From a65fcfc76a0ee0f2d6e5c7b2af6c237645582ae3 Mon Sep 17 00:00:00 2001
From: chantu <lynnleelhl@gmail.com>
Date: Wed, 26 Jul 2023 17:48:16 +0800
Subject: [PATCH] fix: should not check retained backup when deleting cluster
 (#4482)

---
 controllers/apps/cluster_controller_test.go     | 14 ++++++++++++++
 controllers/apps/cluster_plan_builder.go        |  5 -----
 .../apps/transformer_cluster_deletion.go        | 17 +++++++++++++++--
 3 files changed, 29 insertions(+), 7 deletions(-)

diff --git a/controllers/apps/cluster_controller_test.go b/controllers/apps/cluster_controller_test.go
index 143d7e3e265..262a2d03893 100644
--- a/controllers/apps/cluster_controller_test.go
+++ b/controllers/apps/cluster_controller_test.go
@@ -315,6 +315,17 @@ var _ = Describe("Cluster Controller", func() {
 		Eventually(testapps.GetClusterObservedGeneration(&testCtx, clusterKey)).Should(BeEquivalentTo(1))
 		Eventually(testapps.GetClusterPhase(&testCtx, clusterKey)).Should(Equal(appsv1alpha1.CreatingClusterPhase))
 
+		By("Mocking a retained backup")
+		backupPolicyName := "test-backup-policy"
+		backupName := "test-backup"
+		backup := testapps.NewBackupFactory(testCtx.DefaultNamespace, backupName).
+			SetBackupPolicyName(backupPolicyName).
+			SetBackupType(dataprotectionv1alpha1.BackupTypeDataFile).
+			SetLabels(map[string]string{constant.AppInstanceLabelKey: clusterKey.Name, constant.BackupProtectionLabelKey: constant.BackupRetain}).
+			WithRandomName().
+			Create(&testCtx).GetObject()
+		backupKey := client.ObjectKeyFromObject(backup)
+
 		// REVIEW: this test flow
 
 		By("Delete the cluster")
@@ -322,6 +333,9 @@ var _ = Describe("Cluster Controller", func() {
 
 		By("Wait for the cluster to terminate")
 		Eventually(testapps.CheckObjExists(&testCtx, clusterKey, &appsv1alpha1.Cluster{}, false)).Should(Succeed())
+
+		By("Checking backup should exist")
+		Eventually(testapps.CheckObjExists(&testCtx, backupKey, &dataprotectionv1alpha1.Backup{}, true)).Should(Succeed())
 	}
 
 	testDoNotTerminate := func(compName, compDefName string) {
diff --git a/controllers/apps/cluster_plan_builder.go b/controllers/apps/cluster_plan_builder.go
index f57ddb682b8..d20b3e89ada 100644
--- a/controllers/apps/cluster_plan_builder.go
+++ b/controllers/apps/cluster_plan_builder.go
@@ -23,7 +23,6 @@ import (
 	"context"
 	"fmt"
 	"reflect"
-	"strings"
 
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
@@ -265,10 +264,6 @@ func (c *clusterPlanBuilder) reconcileObject(node *ictrltypes.LifecycleVertex) e
 		}
 		// delete secondary objects
 		if _, ok := node.Obj.(*appsv1alpha1.Cluster); !ok {
-			// retain backup for data protection even if the cluster is wiped out.
-			if strings.EqualFold(node.Obj.GetLabels()[constant.BackupProtectionLabelKey], constant.BackupRetain) {
-				return nil
-			}
 			err := intctrlutil.BackgroundDeleteObject(c.cli, c.transCtx.Context, node.Obj)
 			// err := c.cli.Delete(c.transCtx.Context, node.obj)
 			if err != nil && !apierrors.IsNotFound(err) {
diff --git a/controllers/apps/transformer_cluster_deletion.go b/controllers/apps/transformer_cluster_deletion.go
index db1ccbba90f..8d3b248c3ae 100644
--- a/controllers/apps/transformer_cluster_deletion.go
+++ b/controllers/apps/transformer_cluster_deletion.go
@@ -137,18 +137,31 @@ func (t *ClusterDeletionTransformer) Transform(ctx graph.TransformContext, dag *
 		return err
 	}
 
+	toDeleteObjs := func(objs clusterOwningObjects) []client.Object {
+		var delObjs []client.Object
+		for _, obj := range objs {
+			// retain backup for data protection even if the cluster is wiped out.
+			if strings.EqualFold(obj.GetLabels()[constant.BackupProtectionLabelKey], constant.BackupRetain) {
+				continue
+			}
+			delObjs = append(delObjs, obj)
+		}
+		return delObjs
+	}
+
 	// add objects deletion vertex
 	objs, err := getClusterOwningObjects(transCtx, *cluster, ml, toDeleteKinds...)
 	if err != nil {
 		return err
 	}
-	for _, o := range objs {
+	delObjs := toDeleteObjs(objs)
+	for _, o := range delObjs {
 		vertex := &ictrltypes.LifecycleVertex{Obj: o, Action: ictrltypes.ActionDeletePtr()}
 		dag.AddVertex(vertex)
 		dag.Connect(root, vertex)
 	}
 	// set cluster action to noop until all the sub-resources deleted
-	if len(objs) == 0 {
+	if len(delObjs) == 0 {
 		root.Action = ictrltypes.ActionDeletePtr()
 	} else {
 		root.Action = ictrltypes.ActionNoopPtr()