fix(dataprotection): serialize post-ready target jobs

apis/dataprotection/v1alpha1/actionset_types.go

@@ -168,6 +168,14 @@ type RestoreActionSpec struct {
 	// +optional
 	PostReady []ActionSpec `json:"postReady,omitempty"`
 
+	// Specifies how postReady actions run when they target multiple pods.
+	// Parallel preserves the existing behavior. Serial waits for each target
+	// action to complete before starting the next one.
+	//
+	// +optional
+	// +kubebuilder:default=Parallel
+	PostReadyExecutionPolicy PostReadyExecutionPolicy `json:"postReadyExecutionPolicy,omitempty"`
+
 	// Determines if a base backup is required during restoration.
 	//
 	// +optional
@@ -180,6 +188,16 @@ type RestoreActionSpec struct {
 	WithParameters []string `json:"withParameters,omitempty"`
 }
 
+// PostReadyExecutionPolicy specifies how postReady actions execute across target pods.
+// +enum
+// +kubebuilder:validation:Enum={Parallel,Serial}
+type PostReadyExecutionPolicy string
+
+const (
+	PostReadyExecutionPolicyParallel PostReadyExecutionPolicy = "Parallel"
+	PostReadyExecutionPolicySerial   PostReadyExecutionPolicy = "Serial"
+)
+
 // ActionSpec defines an action that should be executed. Only one of the fields may be set.
 type ActionSpec struct {
 	// Specifies that the action should be executed using the pod's exec API within a container.

config/crd/bases/dataprotection.kubeblocks.io_actionsets.yaml

@@ -553,6 +553,16 @@ spec:
                           type: object
                       type: object
                     type: array
+                  postReadyExecutionPolicy:
+                    default: Parallel
+                    description: |-
+                      Specifies how postReady actions run when they target multiple pods.
+                      Parallel preserves the existing behavior. Serial waits for each target
+                      action to complete before starting the next one.
+                    enum:
+                    - Parallel
+                    - Serial
+                    type: string
                   prepareData:
                     description: Specifies the action required to prepare data for
                       restoration.

controllers/dataprotection/restore_controller.go

@@ -62,6 +62,7 @@ type RestoreReconciler struct {
 // +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores/finalizers,verbs=update
 // +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=core,resources=secrets,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=core,resources=serviceaccounts,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebindings,verbs=get;list;watch;create;update;patch;delete
 
@@ -105,6 +106,7 @@ func (r *RestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return intctrlutil.NewControllerManagedBy(mgr).
 		For(&dpv1alpha1.Restore{}).
 		Owns(&batchv1.Job{}).
+		Owns(&corev1.Secret{}).
 		Watches(&batchv1.Job{}, handler.EnqueueRequestsFromMapFunc(r.parseRestoreJob)).
 		// to watch the `restore` container if it is terminated
 		Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(r.parseRestorePod)).
@@ -155,7 +157,11 @@ func (r *RestoreReconciler) deleteExternalResources(reqCtx intctrlutil.RequestCt
 		viper.GetString(constant.CfgKeyCtrlrMgrNS): {},
 	}
 
-	return deleteRelatedObjectList(reqCtx, r.Client, &batchv1.JobList{}, namespaces, labels)
+	if err := deleteRelatedObjectList(reqCtx, r.Client, &batchv1.JobList{}, namespaces, labels); err != nil {
+		return err
+	}
+	return deleteRelatedObjectList(reqCtx, r.Client, &corev1.SecretList{},
+		map[string]sets.Empty{restore.Namespace: {}}, labels)
 }
 
 func CheckBackupRepoForRestore(reqCtx intctrlutil.RequestCtx, cli client.Client, restore *dpv1alpha1.Restore) (string, error) {
@@ -407,37 +413,31 @@ func (r *RestoreReconciler) prepareData(reqCtx intctrlutil.RequestCtx, restoreMg
 }
 
 func (r *RestoreReconciler) postReady(reqCtx intctrlutil.RequestCtx, restoreMgr *dprestore.RestoreManager) (bool, error) {
-	readyConfig := restoreMgr.Restore.Spec.ReadyConfig
-	if len(restoreMgr.PostReadyBackupSets) == 0 || readyConfig == nil {
-		return true, nil
+	foundStage, err := restoreMgr.EnsurePostReadyStagePlan(reqCtx, r.Client)
+	if err != nil {
+		return false, err
+	}
+	if !foundStage {
+		if restoreMgr.Restore.Spec.ReadyConfig == nil || len(restoreMgr.PostReadyBackupSets) == 0 {
+			return true, nil
+		}
+		return false, intctrlutil.NewFatalError("postReady stage has actions but no committed execution plan")
 	}
 	if meta.IsStatusConditionTrue(restoreMgr.Restore.Status.Conditions, dprestore.ConditionTypeRestorePostReady) {
 		return true, nil
 	}
 	dprestore.SetRestoreStageCondition(restoreMgr.Restore, dpv1alpha1.PostReady, dprestore.ReasonProcessing, "processing postReady stage")
-	var (
-		err         error
-		isCompleted bool
-	)
 	defer func() {
 		r.handleRestoreStageError(restoreMgr.Restore, dpv1alpha1.PrepareData, err)
 	}()
-	if readyConfig.ReadinessProbe != nil && !meta.IsStatusConditionTrue(restoreMgr.Restore.Status.Conditions, dprestore.ConditionTypeReadinessProbe) {
+	readyConfig := restoreMgr.Restore.Spec.ReadyConfig
+	if readyConfig != nil && readyConfig.ReadinessProbe != nil && !meta.IsStatusConditionTrue(restoreMgr.Restore.Status.Conditions, dprestore.ConditionTypeReadinessProbe) {
 		// TODO: check readiness probe, use a job and kubectl exec?
 		_ = klog.TODO()
 	}
-	for _, v := range restoreMgr.PostReadyBackupSets {
-		// handle postReady actions
-		for i := range v.ActionSet.Spec.Restore.PostReady {
-			isCompleted, err = r.handleBackupActionSet(reqCtx, restoreMgr, v, dpv1alpha1.PostReady, i)
-			if err != nil {
-				return false, err
-			}
-			// waiting for restore jobs finished.
-			if !isCompleted {
-				return false, nil
-			}
-		}
+	isCompleted, err := restoreMgr.ReconcileOrphanedPostReadyActions(reqCtx, r.Client)
+	if err != nil || !isCompleted {
+		return false, err
 	}
 	dprestore.SetRestoreStageCondition(restoreMgr.Restore, dpv1alpha1.PostReady, dprestore.ReasonSucceed, "processing postReady stage successfully")
 	return true, nil
@@ -469,21 +469,39 @@ func (r *RestoreReconciler) handleBackupActionSet(reqCtx intctrlutil.RequestCtx,
 	}
 
 	actionName := fmt.Sprintf("%s-%d", stage, step)
+	var jobs []*batchv1.Job
+	var err error
+	expectedActionCount := 0
+	if stage == dpv1alpha1.PostReady {
+		// Load the immutable plan before interpreting mutable completion
+		// status. Otherwise one completed ordinal can make a partially
+		// created multi-Job action look complete.
+		jobs, err = restoreMgr.GetExistingActionJobs(reqCtx, r.Client, stage, backupSet.Backup.Name, actionName)
+		if err != nil {
+			return false, err
+		}
+		expectedActionCount, err = dprestore.PostReadyActionExpectedJobCount(jobs)
+		if err != nil {
+			return false, err
+		}
+	}
 	// 1. check if the restore actions are completed from status.actions firstly.
-	allActionsFinished, existFailedAction := restoreMgr.AnalysisRestoreActionsWithBackup(stage, backupSet.Backup.Name, actionName)
+	allActionsFinished, existFailedAction := restoreMgr.AnalysisRestoreActionsWithBackupExpected(
+		stage, backupSet.Backup.Name, actionName, expectedActionCount)
 	isCompleted, err := checkIsCompleted(allActionsFinished, existFailedAction)
 	if isCompleted || err != nil {
 		return isCompleted, err
 	}
 
-	var jobs []*batchv1.Job
 	// For in-flight actions, check the recorded Job before rebuilding the Job
 	// spec from the current target pod selector. The target pod may become
 	// unavailable after the Job is created, but the existing Job is still the
 	// action fact source that should drive convergence.
-	jobs, err = restoreMgr.GetExistingActionJobs(reqCtx, r.Client, stage, backupSet.Backup.Name, actionName)
-	if err != nil {
-		return false, err
+	if stage != dpv1alpha1.PostReady {
+		jobs, err = restoreMgr.GetExistingActionJobs(reqCtx, r.Client, stage, backupSet.Backup.Name, actionName)
+		if err != nil {
+			return false, err
+		}
 	}
 	switch stage {
 	case dpv1alpha1.PrepareData:
@@ -507,11 +525,37 @@ func (r *RestoreReconciler) handleBackupActionSet(reqCtx intctrlutil.RequestCtx,
 	if len(jobs) == 0 {
 		return true, nil
 	}
+	if stage == dpv1alpha1.PostReady {
+		if jobs, err = restoreMgr.FreezePostReadyExecutionPlan(reqCtx, r.Client, jobs); err != nil {
+			return false, err
+		}
+		expectedActionCount, err = dprestore.PostReadyActionExpectedJobCount(jobs)
+		if err != nil {
+			return false, err
+		}
+		allActionsFinished, existFailedAction = restoreMgr.AnalysisRestoreActionsWithBackupExpected(
+			stage, backupSet.Backup.Name, actionName, expectedActionCount)
+		isCompleted, err = checkIsCompleted(allActionsFinished, existFailedAction)
+		if isCompleted || err != nil {
+			return isCompleted, err
+		}
+		jobs = restoreMgr.PendingPostReadyJobs(backupSet.Backup.Name, actionName, jobs)
+		if len(jobs) == 0 {
+			return false, intctrlutil.NewErrorf(intctrlutil.ErrorTypeRequeue,
+				"postReady action %s for backup %s has no pending Job but is not complete",
+				actionName, backupSet.Backup.Name)
+		}
+	}
 	// 3. create jobs
 	jobs, err = restoreMgr.CreateJobsIfNotExist(reqCtx, r.Client, restoreMgr.Restore, jobs)
 	if err != nil {
 		return false, err
 	}
+	if stage == dpv1alpha1.PostReady {
+		if err = restoreMgr.ResumeNextSerialPostReadyJob(reqCtx, r.Client, jobs); err != nil {
+			return false, err
+		}
+	}
 
 	// 4. check if jobs are finished.
 	allActionsFinished, existFailedAction, err = restoreMgr.CheckJobsDone(stage, actionName, backupSet, jobs)

controllers/dataprotection/restore_controller_test.go

@@ -20,6 +20,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package dataprotection
 
 import (
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
@@ -489,6 +490,37 @@ var _ = Describe("Restore Controller test", func() {
 				_ = testdp.NewFakeCluster(&testCtx)
 			})
 
+			expectCommittedFullStage := func(restore *dpv1alpha1.Restore) {
+				Eventually(func(g Gomega) {
+					secrets := &corev1.SecretList{}
+					g.Expect(k8sClient.List(ctx, secrets,
+						client.InNamespace(restore.Namespace),
+						client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name})).Should(Succeed())
+					var payload []byte
+					for i := range secrets.Items {
+						if secrets.Items[i].Type == corev1.SecretType("dataprotection.kubeblocks.io/post-ready-plan") {
+							payload = secrets.Items[i].Data["plan.json"]
+							break
+						}
+					}
+					g.Expect(payload).ShouldNot(BeEmpty())
+					var stage struct {
+						Actions []struct {
+							Order      int    `json:"order"`
+							BackupName string `json:"backupName"`
+							ActionName string `json:"actionName"`
+						} `json:"actions"`
+					}
+					g.Expect(json.Unmarshal(payload, &stage)).Should(Succeed())
+					g.Expect(stage.Actions).Should(HaveLen(2))
+					g.Expect(stage.Actions[0].Order).Should(Equal(0))
+					g.Expect(stage.Actions[0].ActionName).Should(Equal("postReady-0"))
+					g.Expect(stage.Actions[1].Order).Should(Equal(1))
+					g.Expect(stage.Actions[1].ActionName).Should(Equal("postReady-1"))
+					g.Expect(stage.Actions[1].BackupName).Should(Equal(stage.Actions[0].BackupName))
+				}).Should(Succeed())
+			}
+
 			It("test post ready actions", func() {
 				By("remove the prepareData stage for testing post ready actions")
 				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
@@ -507,6 +539,7 @@ var _ = Describe("Restore Controller test", func() {
 				Eventually(testapps.List(&testCtx, generics.JobSignature,
 					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
 					client.InNamespace(testCtx.DefaultNamespace))).Should(HaveLen(2))
+				expectCommittedFullStage(restore)
 
 				checkJobSA(restore, viper.GetString(dptypes.CfgKeyExecWorkerServiceAccountName))
 
@@ -526,6 +559,82 @@ var _ = Describe("Restore Controller test", func() {
 				Eventually(testapps.CheckObjExists(&testCtx, client.ObjectKeyFromObject(restore), restore, false)).Should(Succeed())
 			})
 
+			It("keeps running persisted postReady jobs after the ActionSet removes them", func() {
+				By("remove the prepareData stage for testing post ready actions")
+				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
+					set.Spec.Restore.PrepareData = nil
+				})).Should(Succeed())
+
+				matchLabels := map[string]string{
+					constant.AppInstanceLabelKey: testdp.ClusterName,
+				}
+				restore := initResourcesAndWaitRestore(true, false, false, "", dpv1alpha1.RestorePhaseRunning,
+					func(f *testdp.MockRestoreFactory) {
+						f.SetConnectCredential(testdp.ClusterName).SetJobActionConfig(matchLabels).SetExecActionConfig(matchLabels)
+					}, nil)
+
+				By("wait for the first postReady step to create its Jobs")
+				Eventually(testapps.List(&testCtx, generics.JobSignature,
+					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
+					client.InNamespace(testCtx.DefaultNamespace))).Should(HaveLen(2))
+				expectCommittedFullStage(restore)
+
+				By("remove postReady from the mutable ActionSet while those Jobs are running")
+				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
+					set.Spec.Restore.PostReady = nil
+				})).Should(Succeed())
+				Consistently(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(restore), func(g Gomega, r *dpv1alpha1.Restore) {
+					g.Expect(r.Status.Phase).Should(Equal(dpv1alpha1.RestorePhaseRunning))
+				}), 2*time.Second, 100*time.Millisecond).Should(Succeed())
+
+				By("complete action zero and continue from the frozen action one")
+				mockRestoreJobsCompleted(restore)
+				Eventually(testapps.List(&testCtx, generics.JobSignature,
+					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
+					client.InNamespace(testCtx.DefaultNamespace))).Should(HaveLen(3))
+				mockRestoreJobsCompleted(restore)
+				Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(restore), func(g Gomega, r *dpv1alpha1.Restore) {
+					g.Expect(r.Status.Phase).Should(Equal(dpv1alpha1.RestorePhaseCompleted))
+				})).Should(Succeed())
+			})
+
+			It("finishes a committed postReady plan after the ActionSet is deleted", func() {
+				By("remove the prepareData stage for testing post ready actions")
+				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
+					set.Spec.Restore.PrepareData = nil
+				})).Should(Succeed())
+
+				matchLabels := map[string]string{
+					constant.AppInstanceLabelKey: testdp.ClusterName,
+				}
+				restore := initResourcesAndWaitRestore(true, false, false, "", dpv1alpha1.RestorePhaseRunning,
+					func(f *testdp.MockRestoreFactory) {
+						f.SetConnectCredential(testdp.ClusterName).SetJobActionConfig(matchLabels).SetExecActionConfig(matchLabels)
+					}, nil)
+
+				By("wait for the immutable plan and first postReady Jobs")
+				Eventually(testapps.List(&testCtx, generics.JobSignature,
+					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
+					client.InNamespace(testCtx.DefaultNamespace))).Should(HaveLen(2))
+				expectCommittedFullStage(restore)
+
+				By("delete the mutable ActionSet while the committed Jobs are running")
+				Expect(k8sClient.Delete(ctx, actionSet)).Should(Succeed())
+				Consistently(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(restore), func(g Gomega, r *dpv1alpha1.Restore) {
+					g.Expect(r.Status.Phase).Should(Equal(dpv1alpha1.RestorePhaseRunning))
+				}), 2*time.Second, 100*time.Millisecond).Should(Succeed())
+
+				By("complete action zero and create frozen action one without the deleted ActionSet")
+				mockRestoreJobsCompleted(restore)
+				Eventually(testapps.List(&testCtx, generics.JobSignature,
+					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
+					client.InNamespace(testCtx.DefaultNamespace))).Should(HaveLen(3))
+				mockRestoreJobsCompleted(restore)
+				Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(restore), func(g Gomega, r *dpv1alpha1.Restore) {
+					g.Expect(r.Status.Phase).Should(Equal(dpv1alpha1.RestorePhaseCompleted))
+				})).Should(Succeed())
+			})
+
 			It("should complete an existing postReady job when target pod is no longer ready", func() {
 				By("remove the prepareData stage for testing post ready actions")
 				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {

controllers/dataprotection/utils.go

@@ -302,7 +302,7 @@ func getDefaultBackupRepo(ctx context.Context, cli client.Client) (*dpv1alpha1.B
 }
 
 type objectList interface {
-	*appsv1.StatefulSetList | *batchv1.JobList
+	*appsv1.StatefulSetList | *batchv1.JobList | *corev1.SecretList
 	client.ObjectList
 }
 

deploy/helm/crds/dataprotection.kubeblocks.io_actionsets.yaml

@@ -553,6 +553,16 @@ spec:
                           type: object
                       type: object
                     type: array
+                  postReadyExecutionPolicy:
+                    default: Parallel
+                    description: |-
+                      Specifies how postReady actions run when they target multiple pods.
+                      Parallel preserves the existing behavior. Serial waits for each target
+                      action to complete before starting the next one.
+                    enum:
+                    - Parallel
+                    - Serial
+                    type: string
                   prepareData:
                     description: Specifies the action required to prepare data for
                       restoration.