fix: avoid eager backup delete worker setup (#10343)

Author: weicao

controllers/dataprotection/backup_controller.go

@@ -225,15 +225,11 @@ func (r *BackupReconciler) deleteBackupFiles(reqCtx intctrlutil.RequestCtx, back
 		RequestCtx: reqCtx,
 		Client:     r.Client,
 		Scheme:     r.Scheme,
+		EnsureWorkerServiceAccount: func() (string, error) {
+			return r.ensureWorkerServiceAccountForBackupDeletion(reqCtx, backup.Namespace)
+		},
 	}
 
-	// TODO: update the mcMgr param
-	saName, err := EnsureWorkerServiceAccount(reqCtx, r.Client, backup.Namespace, nil)
-	if err != nil {
-		return fmt.Errorf("failed to get worker service account: %w", err)
-	}
-	deleter.WorkerServiceAccount = saName
-
 	status, err := deleter.DeleteBackupFiles(backup)
 	switch status {
 	case dpbackup.DeletionStatusSucceeded:
@@ -255,6 +251,18 @@ func (r *BackupReconciler) deleteBackupFiles(reqCtx intctrlutil.RequestCtx, back
 	return err
 }
 
+func (r *BackupReconciler) ensureWorkerServiceAccountForBackupDeletion(reqCtx intctrlutil.RequestCtx, namespace string) (string, error) {
+	ns := &corev1.Namespace{}
+	if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: namespace}, ns); err != nil {
+		return "", fmt.Errorf("failed to get backup namespace %q before deleting backup files: %w", namespace, err)
+	}
+	if !ns.DeletionTimestamp.IsZero() {
+		return "", fmt.Errorf("backup namespace %q is terminating; cannot create worker resources to delete backup files, delete the Backup and wait until it is gone before deleting the namespace", namespace)
+	}
+	// TODO: update the mcMgr param
+	return EnsureWorkerServiceAccount(reqCtx, r.Client, namespace, nil)
+}
+
 // handleDeletingPhase handles the deletion of backup. It will delete the backup CR
 // and the backup workload(job).
 func (r *BackupReconciler) handleDeletingPhase(reqCtx intctrlutil.RequestCtx, backup *dpv1alpha1.Backup) (ctrl.Result, error) {

controllers/dataprotection/backup_controller_test.go

@@ -32,9 +32,11 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/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/types"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/pointer"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -789,6 +791,48 @@ var _ = Describe("Backup Controller test", func() {
 
 				// TODO: add delete backup test case with the pvc not exists
 			})
+
+			It("should not create worker resources when backup namespace is terminating", func() {
+				nsName := "backup-delete-terminating"
+				ns := &corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: nsName,
+					},
+				}
+				Expect(client.IgnoreAlreadyExists(testCtx.Cli.Create(testCtx.Ctx, ns))).Should(Succeed())
+
+				finalizeNamespace := func() {
+					Eventually(func(g Gomega) {
+						current := &corev1.Namespace{}
+						err := testCtx.Cli.Get(testCtx.Ctx, types.NamespacedName{Name: nsName}, current)
+						if apierrors.IsNotFound(err) {
+							return
+						}
+						g.Expect(err).ShouldNot(HaveOccurred())
+						current.Spec.Finalizers = []corev1.FinalizerName{}
+						clientGo, err := kubernetes.NewForConfig(testEnv.Config)
+						g.Expect(err).ShouldNot(HaveOccurred())
+						_, err = clientGo.CoreV1().Namespaces().Finalize(testCtx.Ctx, current, metav1.UpdateOptions{})
+						g.Expect(err).ShouldNot(HaveOccurred())
+					}).Should(Succeed())
+				}
+				DeferCleanup(finalizeNamespace)
+
+				By("marking the namespace terminating")
+				Expect(testCtx.Cli.Delete(testCtx.Ctx, ns)).Should(Succeed())
+				Eventually(func(g Gomega) {
+					current := &corev1.Namespace{}
+					g.Expect(testCtx.Cli.Get(testCtx.Ctx, types.NamespacedName{Name: nsName}, current)).Should(Succeed())
+					g.Expect(current.DeletionTimestamp.IsZero()).Should(BeFalse())
+				}).Should(Succeed())
+
+				reconciler := &BackupReconciler{Client: testCtx.Cli}
+				_, err := reconciler.ensureWorkerServiceAccountForBackupDeletion(
+					intctrlutil.RequestCtx{Ctx: testCtx.Ctx},
+					nsName)
+				Expect(err).Should(HaveOccurred())
+				Expect(err.Error()).Should(ContainSubstring("is terminating; cannot create worker resources to delete backup files"))
+			})
 		})
 
 		Context("creates a snapshot backup", func() {

pkg/dataprotection/backup/deleter.go

@@ -59,13 +59,30 @@ const (
 
 type Deleter struct {
 	ctrlutil.RequestCtx
-	Client               client.Client
-	Scheme               *runtime.Scheme
-	WorkerServiceAccount string
+	Client                     client.Client
+	Scheme                     *runtime.Scheme
+	WorkerServiceAccount       string
+	EnsureWorkerServiceAccount func() (string, error)
 
 	actionSet *dpv1alpha1.ActionSet
 }
 
+func (d *Deleter) ensureWorkerServiceAccount() error {
+	if d.WorkerServiceAccount != "" {
+		return nil
+	}
+	if d.EnsureWorkerServiceAccount == nil {
+		d.WorkerServiceAccount = viper.GetString(dptypes.CfgKeyWorkerServiceAccountName)
+		return nil
+	}
+	saName, err := d.EnsureWorkerServiceAccount()
+	if err != nil {
+		return err
+	}
+	d.WorkerServiceAccount = saName
+	return nil
+}
+
 // DeleteBackupFiles builds a job to delete backup files, and returns the deletion status.
 // If the deletion job exists, it will check the job status and return the corresponding
 // deletion status.
@@ -246,6 +263,9 @@ func (d *Deleter) createDeleteJob(container corev1.Container,
 	backup *dpv1alpha1.Backup,
 	backupRepo *dpv1alpha1.BackupRepo,
 	legacyPVCName string) error {
+	if err := d.ensureWorkerServiceAccount(); err != nil {
+		return err
+	}
 	ctrlutil.InjectZeroResourcesLimitsForDataProtection(&container)
 
 	// build pod

pkg/dataprotection/backup/deleter_test.go

@@ -24,6 +24,7 @@ import (
 	. "github.com/onsi/gomega"
 
 	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
@@ -60,6 +61,7 @@ var _ = Describe("Backup Deleter Test", func() {
 		inNS := client.InNamespace(testCtx.DefaultNamespace)
 		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.BackupSignature, true, inNS)
 		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.JobSignature, true, inNS)
+		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.PersistentVolumeClaimSignature, true, inNS)
 		testapps.ClearResources(&testCtx, generics.VolumeSnapshotSignature, inNS)
 	}
 
@@ -107,6 +109,38 @@ var _ = Describe("Backup Deleter Test", func() {
 			Expect(status).Should(Equal(DeletionStatusSucceeded))
 		})
 
+		It("should ensure worker service account only when creating a deletion job", func() {
+			ensureWorkerServiceAccountCalls := 0
+			deleter.EnsureWorkerServiceAccount = func() (string, error) {
+				ensureWorkerServiceAccountCalls++
+				return "worker-sa", nil
+			}
+
+			By("skipping worker service account creation for snapshot backups")
+			backup.Status.BackupMethod = &dpv1alpha1.BackupMethod{
+				SnapshotVolumes: pointer.Bool(true),
+			}
+			status, err := deleter.DeleteBackupFiles(backup)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(status).Should(Equal(DeletionStatusSucceeded))
+			Expect(ensureWorkerServiceAccountCalls).Should(Equal(0))
+
+			By("creating worker service account when a deletion job is needed")
+			backup.Status.BackupMethod = nil
+			backupRepoPVC := testdp.NewFakePVC(&testCtx, backupRepoPVCName)
+			backup.Status.PersistentVolumeClaimName = backupRepoPVC.Name
+			backup.Status.Path = backupPath
+			status, err = deleter.DeleteBackupFiles(backup)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(status).Should(Equal(DeletionStatusDeleting))
+			Expect(ensureWorkerServiceAccountCalls).Should(Equal(1))
+
+			key := BuildDeleteBackupFilesJobKey(backup, false)
+			Eventually(testapps.CheckObj(&testCtx, key, func(g Gomega, fetched *batchv1.Job) {
+				g.Expect(fetched.Spec.Template.Spec.ServiceAccountName).Should(Equal("worker-sa"))
+			})).Should(Succeed())
+		})
+
 		It("should create job to delete backup file", func() {
 			By("mock backup repo PVC")
 			backupRepoPVC := testdp.NewFakePVC(&testCtx, backupRepoPVCName)