fix(dataprotection): terminate cluster restores through resource owners

cmd/dataprotection/main.go

@@ -340,6 +340,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err = (&dpcontrollers.ClusterRestoreReconciler{
+		Client:   mgr.GetClient(),
+		Recorder: mgr.GetEventRecorderFor("cluster-restore-controller"),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "ClusterRestore")
+		os.Exit(1)
+	}
+
 	if err = (&dpcontrollers.BackupScheduleReconciler{
 		Client:   dputils.NewCompatClient(mgr.GetClient()),
 		Scheme:   mgr.GetScheme(),

controllers/apps/cluster/restore_intent.go

@@ -84,6 +84,7 @@ func injectRestoreIntentToVCT(cluster *appsv1.Cluster, componentName string, vct
 	vct.Annotations[constant.RestoreSourceKindAnnotationKey] = restore.Source.Kind
 	vct.Annotations[constant.RestoreSourceNameAnnotationKey] = restore.Source.Name
 	vct.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = sourceNamespace
+	vct.Annotations[constant.KBAppClusterUIDKey] = string(cluster.UID)
 	vct.Annotations[constant.RestoreComponentAnnotationKey] = componentName
 	vct.Annotations[constant.RestoreVolumeTemplateAnnotationKey] = vct.Name
 	delete(vct.Annotations, constant.RestorePITRAnnotationKey)
@@ -118,6 +119,7 @@ func cleanupRestoreIntentFromVCT(vct *appsv1.PersistentVolumeClaimTemplate) {
 		delete(vct.Annotations, constant.RestoreSourceKindAnnotationKey)
 		delete(vct.Annotations, constant.RestoreSourceNameAnnotationKey)
 		delete(vct.Annotations, constant.RestoreSourceNamespaceAnnotationKey)
+		delete(vct.Annotations, constant.KBAppClusterUIDKey)
 		delete(vct.Annotations, constant.RestorePITRAnnotationKey)
 		delete(vct.Annotations, constant.RestoreParametersAnnotationKey)
 		delete(vct.Annotations, constant.RestoreComponentAnnotationKey)
@@ -138,6 +140,7 @@ func hasRestoreIntent(vct *appsv1.PersistentVolumeClaimTemplate) bool {
 		constant.RestoreSourceKindAnnotationKey,
 		constant.RestoreSourceNameAnnotationKey,
 		constant.RestoreSourceNamespaceAnnotationKey,
+		constant.KBAppClusterUIDKey,
 		constant.RestoreComponentAnnotationKey,
 		constant.RestoreVolumeTemplateAnnotationKey,
 	} {

controllers/apps/cluster/restore_intent_test.go

@@ -47,6 +47,7 @@ func TestInjectRestoreIntentRemovesStaleOptionalAnnotations(t *testing.T) {
 	cluster := &appsv1.Cluster{}
 	cluster.Name = "test-cluster"
 	cluster.Namespace = "test-ns"
+	cluster.UID = "cluster-uid"
 	cluster.Spec.Restore = &appsv1.ClusterRestore{
 		Source: appsv1.ClusterRestoreSource{
 			APIGroup:  testRestoreSourceAPIGroup,
@@ -72,6 +73,7 @@ func TestInjectRestoreIntentRemovesStaleOptionalAnnotations(t *testing.T) {
 	require.Equal(t, "backup", vct.Spec.DataSourceRef.Name)
 	require.NotNil(t, vct.Spec.DataSourceRef.Namespace)
 	require.Equal(t, "backup-ns", *vct.Spec.DataSourceRef.Namespace)
+	require.Equal(t, string(cluster.UID), vct.Annotations[constant.KBAppClusterUIDKey])
 }
 
 func TestInjectRestoreIntentOmitsDataSourceRefNamespaceForSameNamespaceSource(t *testing.T) {

controllers/dataprotection/cluster_restore_controller.go

@@ -0,0 +1,189 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package dataprotection
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+	"github.com/apecloud/kubeblocks/pkg/constant"
+	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
+	dprestore "github.com/apecloud/kubeblocks/pkg/dataprotection/restore"
+	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
+)
+
+// ClusterRestoreReconciler coordinates the Cluster-level restore lifecycle.
+type ClusterRestoreReconciler struct {
+	client.Client
+	Recorder record.EventRecorder
+}
+
+// +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=clusters,verbs=get;list;watch;patch;update
+// +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=clusters/finalizers,verbs=update;patch
+// +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores,verbs=get;list;watch
+// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch
+
+func (r *ClusterRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	reqCtx := intctrlutil.RequestCtx{
+		Ctx: ctx, Req: req,
+		Log:      log.FromContext(ctx).WithValues("cluster-restore", req.NamespacedName),
+		Recorder: r.Recorder,
+	}
+	cluster := &appsv1.Cluster{}
+	if err := r.Client.Get(ctx, req.NamespacedName, cluster); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+	}
+
+	hasResources, err := r.hasRestoreResources(ctx, cluster)
+	if err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to inspect Cluster restore resources")
+	}
+	if cluster.DeletionTimestamp.IsZero() {
+		if clusterRestoreConditionActive(cluster) || hasResources {
+			return r.ensureFinalizer(reqCtx, cluster)
+		}
+		return r.removeFinalizer(reqCtx, cluster)
+	}
+	if !controllerutil.ContainsFinalizer(cluster, dptypes.RestoreProtectionFinalizerName) {
+		return intctrlutil.Reconciled()
+	}
+	if hasResources {
+		return intctrlutil.RequeueAfter(reconcileInterval, reqCtx.Log,
+			"waiting for restore resource owners to finish Cluster termination")
+	}
+	return r.removeFinalizer(reqCtx, cluster)
+}
+
+func (r *ClusterRestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return intctrlutil.NewControllerManagedBy(mgr).
+		Named("cluster_restore").
+		For(&appsv1.Cluster{}).
+		Watches(&dpv1alpha1.Restore{}, handler.EnqueueRequestsFromMapFunc(r.mapObjectToCluster)).
+		Watches(&corev1.PersistentVolumeClaim{}, handler.EnqueueRequestsFromMapFunc(r.mapObjectToCluster)).
+		Complete(r)
+}
+
+func (r *ClusterRestoreReconciler) mapObjectToCluster(_ context.Context, obj client.Object) []reconcile.Request {
+	clusterName := obj.GetLabels()[constant.AppInstanceLabelKey]
+	if clusterName == "" {
+		return nil
+	}
+	return []reconcile.Request{{NamespacedName: client.ObjectKey{Namespace: obj.GetNamespace(), Name: clusterName}}}
+}
+
+func clusterRestoreConditionActive(cluster *appsv1.Cluster) bool {
+	if cluster.Spec.Restore == nil {
+		return false
+	}
+	condition := meta.FindStatusCondition(cluster.Status.Conditions, appsv1.ConditionTypeRestore)
+	// Restore=False is terminal for status aggregation, but the failed Cluster
+	// still carries initial-restore intent. Keep the protection until the user
+	// deletes the Cluster so remaining PVC restores never observe an absent
+	// protection finalizer and stall between App and DP state machines.
+	return condition == nil || condition.Status != metav1.ConditionTrue
+}
+
+func (r *ClusterRestoreReconciler) hasRestoreResources(ctx context.Context,
+	cluster *appsv1.Cluster) (bool, error) {
+	restores := &dpv1alpha1.RestoreList{}
+	if err := r.Client.List(ctx, restores, client.InNamespace(cluster.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: cluster.Name,
+	}); err != nil {
+		return false, err
+	}
+	for i := range restores.Items {
+		restore := &restores.Items[i]
+		if restore.Labels[dprestore.DataProtectionRestoreLabelKey] != restore.Name {
+			continue
+		}
+		owned := restore.Labels[dptypes.ClusterUIDLabelKey] == string(cluster.UID)
+		terminal := restore.Status.Phase == dpv1alpha1.RestorePhaseCompleted ||
+			restore.Status.Phase == dpv1alpha1.RestorePhaseFailed
+		if owned && (!cluster.DeletionTimestamp.IsZero() ||
+			!terminal || !restore.DeletionTimestamp.IsZero()) {
+			return true, nil
+		}
+	}
+
+	pvcs := &corev1.PersistentVolumeClaimList{}
+	if err := r.Client.List(ctx, pvcs, client.InNamespace(cluster.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: cluster.Name,
+	}); err != nil {
+		return false, err
+	}
+	for i := range pvcs.Items {
+		pvc := &pvcs.Items[i]
+		if pvc.Labels[dptypes.ClusterUIDLabelKey] != string(cluster.UID) {
+			continue
+		}
+		if isClusterRestoreHelperPVC(pvc) ||
+			(isClusterRestoreTargetPVC(pvc) && controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName)) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func (r *ClusterRestoreReconciler) ensureFinalizer(reqCtx intctrlutil.RequestCtx,
+	cluster *appsv1.Cluster) (ctrl.Result, error) {
+	if controllerutil.ContainsFinalizer(cluster, dptypes.RestoreProtectionFinalizerName) {
+		return intctrlutil.Reconciled()
+	}
+	patch := client.MergeFromWithOptions(cluster.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	controllerutil.AddFinalizer(cluster, dptypes.RestoreProtectionFinalizerName)
+	if err := r.Client.Patch(reqCtx.Ctx, cluster, patch); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to add Cluster restore-protection finalizer")
+	}
+	return intctrlutil.Reconciled()
+}
+
+func (r *ClusterRestoreReconciler) removeFinalizer(reqCtx intctrlutil.RequestCtx,
+	cluster *appsv1.Cluster) (ctrl.Result, error) {
+	if !controllerutil.ContainsFinalizer(cluster, dptypes.RestoreProtectionFinalizerName) {
+		return intctrlutil.Reconciled()
+	}
+	patch := client.MergeFromWithOptions(cluster.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	controllerutil.RemoveFinalizer(cluster, dptypes.RestoreProtectionFinalizerName)
+	if err := r.Client.Patch(reqCtx.Ctx, cluster, patch); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to remove Cluster restore-protection finalizer")
+	}
+	return intctrlutil.Reconciled()
+}
+
+func isClusterRestoreHelperPVC(pvc *corev1.PersistentVolumeClaim) bool {
+	return pvc.Labels[dprestore.DataProtectionPopulatePVCLabelKey] != ""
+}
+
+func isClusterRestoreTargetPVC(pvc *corev1.PersistentVolumeClaim) bool {
+	return pvc.Spec.DataSourceRef != nil && pvc.Spec.DataSourceRef.APIGroup != nil &&
+		*pvc.Spec.DataSourceRef.APIGroup == dptypes.DataprotectionAPIGroup
+}