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

@@ -512,6 +512,11 @@ func (r *RestoreReconciler) handleBackupActionSet(reqCtx intctrlutil.RequestCtx,
 	if err != nil {
 		return false, err
 	}
+	if stage == dpv1alpha1.PostReady {
+		if err = restoreMgr.ResumeNextSerialPostReadyJob(reqCtx, r.Client, backupSet, jobs); err != nil {
+			return false, err
+		}
+	}
 
 	// 4. check if jobs are finished.
 	allActionsFinished, existFailedAction, err = restoreMgr.CheckJobsDone(stage, actionName, backupSet, jobs)

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.

docs/developer_docs/api-reference/dataprotection.md

@@ -4796,6 +4796,27 @@ And only takes effect when the ‘strategy’ is set to ‘Any&rsquo
 </tr>
 </tbody>
 </table>
+<h3 id="dataprotection.kubeblocks.io/v1alpha1.PostReadyExecutionPolicy">PostReadyExecutionPolicy
+(<code>string</code> alias)</h3>
+<p>
+(<em>Appears on:</em><a href="#dataprotection.kubeblocks.io/v1alpha1.RestoreActionSpec">RestoreActionSpec</a>)
+</p>
+<div>
+<p>PostReadyExecutionPolicy specifies how postReady actions execute across target pods.</p>
+</div>
+<table>
+<thead>
+<tr>
+<th>Value</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody><tr><td><p>&#34;Parallel&#34;</p></td>
+<td></td>
+</tr><tr><td><p>&#34;Serial&#34;</p></td>
+<td></td>
+</tr></tbody>
+</table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.PrepareDataConfig">PrepareDataConfig
 </h3>
 <p>
@@ -5180,6 +5201,22 @@ JobActionSpec
 </tr>
 <tr>
 <td>
+<code>postReadyExecutionPolicy</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.PostReadyExecutionPolicy">
+PostReadyExecutionPolicy
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>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.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>baseBackupRequired</code><br/>
 <em>
 bool

pkg/dataprotection/restore/manager.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -800,10 +801,75 @@ func (r *RestoreManager) BuildPostReadyActionJobs(reqCtx intctrlutil.RequestCtx,
 		return restoreJobs, nil
 	}
 
+	var jobs []*batchv1.Job
 	if actionSpec.Job != nil {
-		return buildJobsForJobAction()
+		jobs, err = buildJobsForJobAction()
+	} else {
+		jobs, err = buildJobsForExecAction()
+	}
+	if err != nil {
+		return nil, err
+	}
+	if isSerialPostReady(backupSet) {
+		for i := range jobs {
+			suspend := i > 0
+			jobs[i].Spec.Suspend = &suspend
+		}
+	}
+	return jobs, nil
+}
+
+func isSerialPostReady(backupSet BackupActionSet) bool {
+	return backupSet.ActionSet != nil &&
+		backupSet.ActionSet.Spec.Restore != nil &&
+		backupSet.ActionSet.Spec.Restore.PostReadyExecutionPolicy == dpv1alpha1.PostReadyExecutionPolicySerial
+}
+
+// ResumeNextSerialPostReadyJob starts at most one suspended postReady job after
+// every preceding job has completed successfully.
+func (r *RestoreManager) ResumeNextSerialPostReadyJob(
+	reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
+	backupSet BackupActionSet,
+	jobs []*batchv1.Job,
+) error {
+	if !isSerialPostReady(backupSet) {
+		return nil
+	}
+	sort.Slice(jobs, func(i, j int) bool {
+		return postReadyJobIndex(jobs[i].Name) < postReadyJobIndex(jobs[j].Name)
+	})
+	for i := range jobs {
+		done, _, errMsg := utils.IsJobFinished(jobs[i])
+		if errMsg != "" {
+			return nil
+		}
+		if done {
+			continue
+		}
+		if jobs[i].Spec.Suspend != nil && *jobs[i].Spec.Suspend {
+			updated := jobs[i].DeepCopy()
+			updated.Spec.Suspend = boolptr.False()
+			if err := cli.Patch(reqCtx.Ctx, updated, client.MergeFrom(jobs[i])); err != nil {
+				return err
+			}
+			jobs[i] = updated
+		}
+		return nil
 	}
-	return buildJobsForExecAction()
+	return nil
+}
+
+func postReadyJobIndex(name string) int {
+	separator := strings.LastIndexByte(name, '-')
+	if separator < 0 {
+		return int(^uint(0) >> 1)
+	}
+	index, err := strconv.Atoi(name[separator+1:])
+	if err != nil {
+		return int(^uint(0) >> 1)
+	}
+	return index
 }
 
 func (r *RestoreManager) createPVCIfNotExist(
@@ -883,6 +949,7 @@ func (r *RestoreManager) CheckJobsDone(
 	if stage == dpv1alpha1.PostReady {
 		restoreActions = &r.Restore.Status.Actions.PostReady
 	}
+	serialPostReady := stage == dpv1alpha1.PostReady && isSerialPostReady(backupSet)
 	// count the number of jobs that are completed, failed,
 	// or have the normally terminated `restore` container
 	finishedCount := 0
@@ -914,9 +981,17 @@ func (r *RestoreManager) CheckJobsDone(
 			}
 			if normalTerminated {
 				finishedCount++
+				if serialPostReady {
+					if err := r.StopManagerContainerByJob(fetchedJobs[i]); err != nil {
+						return false, false, err
+					}
+				}
 			}
 		}
 	}
+	if serialPostReady && existFailedJob {
+		return true, true, nil
+	}
 	// wait until all `restore` containers are terminated normally or jobs are completed or failed
 	if finishedCount == len(fetchedJobs) {
 		for i := range fetchedJobs {

pkg/dataprotection/restore/manager_test.go

@@ -499,6 +499,9 @@ var _ = Describe("RestoreManager Test", func() {
 			Expect(err).ShouldNot(HaveOccurred())
 			// the count of exec jobs should equal to the pods count of cluster
 			Expect(len(jobs)).Should(Equal(2))
+			for i := range jobs {
+				Expect(jobs[i].Spec.Suspend).Should(BeNil())
+			}
 			Expect(jobs[0].Namespace).Should(Equal(kbNamespace))
 			Expect(jobs[0].Spec.Template.Spec.ServiceAccountName).Should(Equal(execWorkerServiceAccountName))
 
@@ -532,6 +535,52 @@ var _ = Describe("RestoreManager Test", func() {
 			testPostReady(false)
 		})
 
+		It("serializes postReady jobs across selected pods", func() {
+			reqCtx := getReqCtx()
+			matchLabels := map[string]string{
+				constant.AppInstanceLabelKey: testdp.ClusterName,
+			}
+			Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
+				set.Spec.Restore.PostReadyExecutionPolicy = dpv1alpha1.PostReadyExecutionPolicySerial
+			})).Should(Succeed())
+			oldControllerNamespace := viper.GetString(constant.CfgKeyCtrlrMgrNS)
+			viper.Set(constant.CfgKeyCtrlrMgrNS, testCtx.DefaultNamespace)
+			DeferCleanup(func() { viper.Set(constant.CfgKeyCtrlrMgrNS, oldControllerNamespace) })
+			restoreMGR, backupSet := initResources(reqCtx, 0, false, func(f *testdp.MockRestoreFactory) {
+				f.SetConnectCredential(testdp.ClusterName).SetJobActionConfig(matchLabels).SetExecActionConfig(matchLabels)
+			})
+			testdp.NewFakeCluster(&testCtx)
+
+			target := utils.GetBackupStatusTarget(backupSet.Backup, restoreMGR.Restore.Spec.Backup.SourceTargetName)
+			jobs, err := restoreMGR.BuildPostReadyActionJobs(reqCtx, k8sClient, *backupSet, target, 0)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(jobs).Should(HaveLen(2))
+			Expect(jobs[0].Spec.Suspend).ShouldNot(BeNil())
+			Expect(*jobs[0].Spec.Suspend).Should(BeFalse())
+			Expect(jobs[1].Spec.Suspend).ShouldNot(BeNil())
+			Expect(*jobs[1].Spec.Suspend).Should(BeTrue())
+			for i := range jobs {
+				for j := range jobs[i].Spec.Template.Spec.Containers {
+					jobs[i].Spec.Template.Spec.Containers[j].Image = "test-image"
+				}
+			}
+
+			jobs, err = restoreMGR.CreateJobsIfNotExist(reqCtx, k8sClient, restoreMGR.Restore, jobs)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(restoreMGR.ResumeNextSerialPostReadyJob(reqCtx, k8sClient, *backupSet, jobs)).Should(Succeed())
+			second := &batchv1.Job{}
+			Expect(k8sClient.Get(reqCtx.Ctx, client.ObjectKeyFromObject(jobs[1]), second)).Should(Succeed())
+			Expect(*second.Spec.Suspend).Should(BeTrue())
+
+			testdp.PatchK8sJobStatus(&testCtx, client.ObjectKeyFromObject(jobs[0]), batchv1.JobComplete)
+			for i := range jobs {
+				Expect(k8sClient.Get(reqCtx.Ctx, client.ObjectKeyFromObject(jobs[i]), jobs[i])).Should(Succeed())
+			}
+			Expect(restoreMGR.ResumeNextSerialPostReadyJob(reqCtx, k8sClient, *backupSet, jobs)).Should(Succeed())
+			Expect(k8sClient.Get(reqCtx.Ctx, client.ObjectKeyFromObject(jobs[1]), second)).Should(Succeed())
+			Expect(*second.Spec.Suspend).Should(BeFalse())
+		})
+
 		Context("BuildContinuousRestoreManager", func() {
 			const (
 				continuousBackupStartTime = "2023-01-01T09:00:00Z"