From fb929e26d737eecc19e81c640eb90fdcaf8f1733 Mon Sep 17 00:00:00 2001
From: zjx20 <x.zhou@apecloud.com>
Date: Mon, 16 Oct 2023 11:51:25 +0800
Subject: [PATCH] feat: switch to using the datasafed tool for backup/restore
 scripts (#5451)

(cherry picked from commit 59e81b5601b14c4bf1da3a39c4c6ce45bf8e8f38)
---
 .../v1alpha1/backuprepo_types.go              |   1 +
 ...aprotection.kubeblocks.io_backuprepos.yaml |   3 +
 .../dataprotection/backup_controller.go       |  56 +++++--
 .../dataprotection/backup_controller_test.go  |   3 +-
 .../dataprotection/backuprepo_controller.go   |  16 +-
 .../backuprepo_controller_test.go             |  44 +++++-
 .../volumepopulator_controller.go             |   2 +-
 .../apecloud-mysql/dataprotection/backup.sh   |  15 +-
 .../apecloud-mysql/dataprotection/restore.sh  |   5 +-
 ...aprotection.kubeblocks.io_backuprepos.yaml |   3 +
 .../helm/templates/addons/csi-s3-addon.yaml   |   6 +-
 deploy/helm/templates/dataprotection.yaml     |   2 +
 .../helm/templates/storageprovider/cos.yaml   |   1 +
 .../helm/templates/storageprovider/gcs.yaml   |   1 +
 .../helm/templates/storageprovider/minio.yaml |   1 +
 .../helm/templates/storageprovider/obs.yaml   |   1 +
 .../helm/templates/storageprovider/oss.yaml   |   1 +
 deploy/helm/templates/storageprovider/s3.yaml |   1 +
 deploy/helm/values.yaml                       |   5 +
 .../dataprotection/backup-info-collector.sh   |   6 +-
 .../mongodb/dataprotection/datafile-backup.sh |  11 +-
 .../dataprotection/datafile-restore.sh        |   5 +-
 .../dataprotection/mongodump-backup.sh        |  10 +-
 .../dataprotection/mongodump-restore.sh       |  11 +-
 deploy/mongodb/dataprotection/pitr-backup.sh  |  83 -----------
 deploy/mongodb/scripts/replicaset-restore.tpl |  29 ----
 .../templates/actionset-xtrabackup.yaml       |  19 ++-
 .../dataprotection/backup-info-collector.sh   |   6 +-
 .../dataprotection/pg-basebackup-backup.sh    |  11 +-
 .../dataprotection/pg-basebackup-restore.sh   |  18 +--
 .../templates/actionset-pgbasebackup.yaml     |   2 -
 deploy/qdrant/scripts/qdrant-backup.sh        |  13 +-
 deploy/qdrant/scripts/qdrant-restore.sh       |  14 +-
 deploy/redis/dataprotection/backup.sh         |  11 +-
 deploy/redis/dataprotection/restore.sh        |   5 +-
 deploy/redis/templates/backupactionset.yaml   |   2 +-
 docs/user_docs/cli/cli.md                     |   4 +-
 docs/user_docs/cli/kbcli_backuprepo.md        |   4 +-
 docs/user_docs/cli/kbcli_backuprepo_create.md |   1 +
 .../cli/kbcli_backuprepo_describe.md          |   2 +-
 docs/user_docs/cli/kbcli_backuprepo_list.md   |   4 +-
 internal/cli/cmd/backuprepo/create.go         |  60 +++++++-
 internal/cli/cmd/backuprepo/create_test.go    |  46 +++++-
 internal/cli/cmd/backuprepo/describe.go       |  15 +-
 internal/cli/cmd/backuprepo/list.go           |   9 +-
 internal/cli/cmd/cluster/dataprotection.go    |   7 +-
 internal/cli/testing/fake.go                  |   5 +-
 internal/dataprotection/backup/deleter.go     |  98 +++++++------
 internal/dataprotection/backup/request.go     |  69 +++++----
 internal/dataprotection/backup/utils.go       |  30 ----
 internal/dataprotection/errors/errors.go      |  14 ++
 internal/dataprotection/errors/errors_test.go |  12 +-
 internal/dataprotection/restore/builder.go    |  43 +++---
 internal/dataprotection/restore/manager.go    |  43 +++++-
 .../dataprotection/restore/manager_test.go    |   2 +-
 internal/dataprotection/types/constant.go     |   8 +-
 internal/dataprotection/utils/backuprepo.go   | 137 ++++++++++++++++++
 .../testutil/dataprotection/backup_utils.go   |   7 +
 58 files changed, 673 insertions(+), 370 deletions(-)
 delete mode 100644 deploy/mongodb/dataprotection/pitr-backup.sh
 delete mode 100644 deploy/mongodb/scripts/replicaset-restore.tpl
 create mode 100644 internal/dataprotection/utils/backuprepo.go

diff --git a/apis/dataprotection/v1alpha1/backuprepo_types.go b/apis/dataprotection/v1alpha1/backuprepo_types.go
index f89c6156e35..aa4a46f666a 100644
--- a/apis/dataprotection/v1alpha1/backuprepo_types.go
+++ b/apis/dataprotection/v1alpha1/backuprepo_types.go
@@ -109,6 +109,7 @@ type BackupRepoStatus struct {
 // +kubebuilder:resource:path=backuprepos,categories={kubeblocks},scope=Cluster
 // +kubebuilder:printcolumn:name="STATUS",type="string",JSONPath=".status.phase"
 // +kubebuilder:printcolumn:name="STORAGEPROVIDER",type="string",JSONPath=".spec.storageProviderRef"
+// +kubebuilder:printcolumn:name="ACCESSMETHOD",type="string",JSONPath=".spec.accessMethod"
 // +kubebuilder:printcolumn:name="DEFAULT",type="boolean",JSONPath=`.status.isDefault`
 // +kubebuilder:printcolumn:name="AGE",type="date",JSONPath=".metadata.creationTimestamp"
 
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_backuprepos.yaml b/config/crd/bases/dataprotection.kubeblocks.io_backuprepos.yaml
index 00fcc2ab7a4..43188e28b3d 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_backuprepos.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_backuprepos.yaml
@@ -24,6 +24,9 @@ spec:
     - jsonPath: .spec.storageProviderRef
       name: STORAGEPROVIDER
       type: string
+    - jsonPath: .spec.accessMethod
+      name: ACCESSMETHOD
+      type: string
     - jsonPath: .status.isDefault
       name: DEFAULT
       type: boolean
diff --git a/controllers/dataprotection/backup_controller.go b/controllers/dataprotection/backup_controller.go
index 280029e1ffa..049559c66af 100644
--- a/controllers/dataprotection/backup_controller.go
+++ b/controllers/dataprotection/backup_controller.go
@@ -302,20 +302,42 @@ func (r *BackupReconciler) handleBackupRepo(request *dpbackup.Request) error {
 	}
 	request.BackupRepo = repo
 
-	pvcName := repo.Status.BackupPVCName
-	if pvcName == "" {
-		return dperrors.NewBackupPVCNameIsEmpty(repo.Name, request.Spec.BackupPolicyName)
+	if repo.Status.Phase != dpv1alpha1.BackupRepoReady {
+		return dperrors.NewBackupRepoIsNotReady(repo.Name)
 	}
 
-	pvc := &corev1.PersistentVolumeClaim{}
-	pvcKey := client.ObjectKey{Namespace: request.Req.Namespace, Name: pvcName}
-	if err = r.Client.Get(request.Ctx, pvcKey, pvc); err != nil {
-		return client.IgnoreNotFound(err)
-	}
-
-	// backupRepo PVC exists, record the PVC name
-	if err == nil {
-		request.BackupRepoPVC = pvc
+	switch {
+	case repo.AccessByMount():
+		pvcName := repo.Status.BackupPVCName
+		if pvcName == "" {
+			return dperrors.NewBackupPVCNameIsEmpty(repo.Name, request.Spec.BackupPolicyName)
+		}
+		pvc := &corev1.PersistentVolumeClaim{}
+		pvcKey := client.ObjectKey{Namespace: request.Req.Namespace, Name: pvcName}
+		if err = r.Client.Get(request.Ctx, pvcKey, pvc); err != nil {
+			// will wait for the backuprepo controller to create the PVC,
+			// so ignore the NotFound error
+			return client.IgnoreNotFound(err)
+		}
+		// backupRepo PVC exists, record the PVC name
+		if err == nil {
+			request.BackupRepoPVC = pvc
+		}
+	case repo.AccessByTool():
+		toolConfigSecretName := repo.Status.ToolConfigSecretName
+		if toolConfigSecretName == "" {
+			return dperrors.NewToolConfigSecretNameIsEmpty(repo.Name)
+		}
+		secret := &corev1.Secret{}
+		secretKey := client.ObjectKey{Namespace: request.Req.Namespace, Name: toolConfigSecretName}
+		if err = r.Client.Get(request.Ctx, secretKey, secret); err != nil {
+			// will wait for the backuprepo controller to create the secret,
+			// so ignore the NotFound error
+			return client.IgnoreNotFound(err)
+		}
+		if err == nil {
+			request.ToolConfigSecret = secret
+		}
 	}
 	return nil
 }
@@ -327,8 +349,10 @@ func (r *BackupReconciler) patchBackupStatus(
 	request.Status.Path = dpbackup.BuildBackupPath(request.Backup, request.BackupPolicy.Spec.PathPrefix)
 	request.Status.Target = request.BackupPolicy.Spec.Target
 	request.Status.BackupMethod = request.BackupMethod
-	request.Status.PersistentVolumeClaimName = request.BackupRepoPVC.Name
 	request.Status.BackupRepoName = request.BackupRepo.Name
+	if request.BackupRepoPVC != nil {
+		request.Status.PersistentVolumeClaimName = request.BackupRepoPVC.Name
+	}
 
 	// init action status
 	actions, err := request.BuildActions()
@@ -383,10 +407,10 @@ func (r *BackupReconciler) patchBackupObjectMeta(
 	request.Labels[constant.AppManagedByLabelKey] = constant.AppName
 	request.Labels[dataProtectionLabelBackupTypeKey] = request.GetBackupType()
 
-	// if the backupRepo PVC is not present, add a special label and wait for the
-	// backup repo controller to create the PVC.
+	// wait for the backup repo controller to prepare the essential resource.
 	wait := false
-	if request.BackupRepoPVC == nil {
+	if (request.BackupRepo.AccessByMount() && request.BackupRepoPVC == nil) ||
+		(request.BackupRepo.AccessByTool() && request.ToolConfigSecret == nil) {
 		request.Labels[dataProtectionWaitRepoPreparationKey] = trueVal
 		wait = true
 	}
diff --git a/controllers/dataprotection/backup_controller_test.go b/controllers/dataprotection/backup_controller_test.go
index 066bec6c2ef..ea9b8dc235c 100644
--- a/controllers/dataprotection/backup_controller_test.go
+++ b/controllers/dataprotection/backup_controller_test.go
@@ -208,7 +208,7 @@ var _ = Describe("Backup Controller test", func() {
 				jobKey := dpbackup.BuildDeleteBackupFilesJobKey(backup)
 				job := &batchv1.Job{}
 				Eventually(testapps.CheckObjExists(&testCtx, jobKey, job, true)).Should(Succeed())
-				volumeName := dpbackup.GenerateBackupRepoVolumeName(repoPVCName)
+				volumeName := "dp-backup-data"
 				Eventually(testapps.CheckObj(&testCtx, jobKey, func(g Gomega, job *batchv1.Job) {
 					Expect(job.Spec.Template.Spec.Volumes).
 						Should(ContainElement(corev1.Volume{
@@ -476,6 +476,7 @@ var _ = Describe("Backup Controller test", func() {
 					Eventually(testapps.GetAndChangeObjStatus(&testCtx, client.ObjectKeyFromObject(sp),
 						func(fetched *storagev1alpha1.StorageProvider) {
 							fetched.Status.Phase = storagev1alpha1.StorageProviderNotReady
+							fetched.Status.Conditions = nil
 						})).ShouldNot(HaveOccurred())
 					Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(repo),
 						func(g Gomega, repo *dpv1alpha1.BackupRepo) {
diff --git a/controllers/dataprotection/backuprepo_controller.go b/controllers/dataprotection/backuprepo_controller.go
index 69b547453cb..bc621ddc76e 100644
--- a/controllers/dataprotection/backuprepo_controller.go
+++ b/controllers/dataprotection/backuprepo_controller.go
@@ -266,6 +266,11 @@ func (r *BackupRepoReconciler) checkStorageProvider(
 			reason = ReasonInvalidStorageProvider
 			return provider, newDependencyError("both StorageClassTemplate and PersistentVolumeClaimTemplate are empty")
 		}
+		csiInstalledCond := meta.FindStatusCondition(provider.Status.Conditions, storagev1alpha1.ConditionTypeCSIDriverInstalled)
+		if csiInstalledCond == nil || csiInstalledCond.Status != metav1.ConditionTrue {
+			reason = ReasonStorageProviderNotReady
+			return provider, newDependencyError("CSI driver is not installed")
+		}
 	case repo.AccessByTool():
 		if provider.Spec.DatasafedConfigTemplate == "" {
 			reason = ReasonInvalidStorageProvider
@@ -274,15 +279,8 @@ func (r *BackupRepoReconciler) checkStorageProvider(
 	}
 
 	// check its status
-	if provider.Status.Phase == storagev1alpha1.StorageProviderReady {
-		reason = ReasonStorageProviderReady
-		return provider, nil
-	} else {
-		reason = ReasonStorageProviderNotReady
-		err = newDependencyError(fmt.Sprintf("storage provider %s is not ready, status: %s",
-			provider.Name, provider.Status.Phase))
-		return provider, err
-	}
+	reason = ReasonStorageProviderReady
+	return provider, nil
 }
 
 func (r *BackupRepoReconciler) checkParameters(reqCtx intctrlutil.RequestCtx,
diff --git a/controllers/dataprotection/backuprepo_controller_test.go b/controllers/dataprotection/backuprepo_controller_test.go
index fdee074a326..be5190c950c 100644
--- a/controllers/dataprotection/backuprepo_controller_test.go
+++ b/controllers/dataprotection/backuprepo_controller_test.go
@@ -185,6 +185,11 @@ parameters:
     secret-namespace: {{ .CSIDriverSecretRef.Namespace }}
 `
 			obj.Status.Phase = storagev1alpha1.StorageProviderReady
+			meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{
+				Type:   storagev1alpha1.ConditionTypeCSIDriverInstalled,
+				Status: metav1.ConditionTrue,
+				Reason: "CSIDriverInstalled",
+			})
 			if mutateFunc != nil {
 				mutateFunc(obj)
 			}
@@ -306,6 +311,11 @@ parameters:
 			By("updating the status of the storage provider to not ready")
 			Eventually(testapps.GetAndChangeObjStatus(&testCtx, providerKey, func(provider *storagev1alpha1.StorageProvider) {
 				provider.Status.Phase = storagev1alpha1.StorageProviderNotReady
+				meta.SetStatusCondition(&provider.Status.Conditions, metav1.Condition{
+					Type:   storagev1alpha1.ConditionTypeCSIDriverInstalled,
+					Status: metav1.ConditionFalse,
+					Reason: "CSINotInstalled",
+				})
 			})).Should(Succeed())
 			By("checking the status of the BackupRepo, should become failed")
 			Eventually(testapps.CheckObj(&testCtx, repoKey, func(g Gomega, repo *dpv1alpha1.BackupRepo) {
@@ -664,8 +674,7 @@ spec:
 			var backup *dpv1alpha1.Backup
 			var toolConfigSecretKey types.NamespacedName
 
-			BeforeEach(func() {
-				By("preparing")
+			createStorageProviderSpecForToolAccessMethod := func(mutateFunc func(provider *storagev1alpha1.StorageProvider)) {
 				createStorageProviderSpec(func(provider *storagev1alpha1.StorageProvider) {
 					provider.Spec.DatasafedConfigTemplate = `
 [storage]
@@ -675,7 +684,15 @@ key2={{ index .Parameters "key2" }}
 cred-key1={{ index .Parameters "cred-key1" }}
 cred-key2={{ index .Parameters "cred-key2" }}
 `
+					if mutateFunc != nil {
+						mutateFunc(provider)
+					}
 				})
+			}
+
+			BeforeEach(func() {
+				By("preparing")
+				createStorageProviderSpecForToolAccessMethod(nil)
 				createBackupRepoSpec(func(repo *dpv1alpha1.BackupRepo) {
 					repo.Spec.AccessMethod = dpv1alpha1.AccessMethodTool
 				})
@@ -695,7 +712,7 @@ cred-key2={{ index .Parameters "cred-key2" }}
 
 			It("should check that the storage provider has a non-empty datasafedConfigTemplate", func() {
 				By("preparing")
-				createStorageProviderSpec(func(provider *storagev1alpha1.StorageProvider) {
+				createStorageProviderSpecForToolAccessMethod(func(provider *storagev1alpha1.StorageProvider) {
 					provider.Spec.DatasafedConfigTemplate = ""
 				})
 				createBackupRepoSpec(func(repo *dpv1alpha1.BackupRepo) {
@@ -714,7 +731,7 @@ cred-key2={{ index .Parameters "cred-key2" }}
 
 			It("should fail if the datasafedConfigTemplate is invalid", func() {
 				By("preparing")
-				createStorageProviderSpec(func(provider *storagev1alpha1.StorageProvider) {
+				createStorageProviderSpecForToolAccessMethod(func(provider *storagev1alpha1.StorageProvider) {
 					provider.Spec.DatasafedConfigTemplate = "bad template {{"
 				})
 				createBackupRepoSpec(func(repo *dpv1alpha1.BackupRepo) {
@@ -730,6 +747,25 @@ cred-key2={{ index .Parameters "cred-key2" }}
 				})).Should(Succeed())
 			})
 
+			It("should work even if the CSI driver required by the storage provider is not installed", func() {
+				By("preparing")
+				createStorageProviderSpecForToolAccessMethod(func(provider *storagev1alpha1.StorageProvider) {
+					provider.Status.Phase = storagev1alpha1.StorageProviderNotReady
+					meta.SetStatusCondition(&provider.Status.Conditions, metav1.Condition{
+						Type:   storagev1alpha1.ConditionTypeCSIDriverInstalled,
+						Status: metav1.ConditionFalse,
+						Reason: "NotInstalled",
+					})
+				})
+				createBackupRepoSpec(func(repo *dpv1alpha1.BackupRepo) {
+					repo.Spec.AccessMethod = dpv1alpha1.AccessMethodTool
+				})
+				By("checking")
+				Eventually(testapps.CheckObj(&testCtx, repoKey, func(g Gomega, repo *dpv1alpha1.BackupRepo) {
+					g.Expect(repo.Status.Phase).Should(Equal(dpv1alpha1.BackupRepoReady))
+				})).Should(Succeed())
+			})
+
 			It("should create the secret containing the tool config", func() {
 				Eventually(testapps.CheckObj(&testCtx, toolConfigSecretKey, func(g Gomega, secret *corev1.Secret) {
 					g.Expect(secret.Data).Should(HaveKeyWithValue("datasafed.conf", []byte(`
diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go
index 621adf24deb..53980c067a3 100644
--- a/controllers/dataprotection/volumepopulator_controller.go
+++ b/controllers/dataprotection/volumepopulator_controller.go
@@ -173,7 +173,7 @@ func (r *VolumePopulatorReconciler) populate(reqCtx intctrlutil.RequestCtx, pvc
 		}
 
 		// 1. build populate job
-		job, err := restoreMgr.BuildVolumePopulateJob(v, populatePVC, i)
+		job, err := restoreMgr.BuildVolumePopulateJob(reqCtx, r.Client, v, populatePVC, i)
 		if err != nil {
 			return err
 		}
diff --git a/deploy/apecloud-mysql/dataprotection/backup.sh b/deploy/apecloud-mysql/dataprotection/backup.sh
index 507a214cbab..ab134f4636e 100644
--- a/deploy/apecloud-mysql/dataprotection/backup.sh
+++ b/deploy/apecloud-mysql/dataprotection/backup.sh
@@ -1,12 +1,13 @@
 #!/bin/bash
 set -e
-if [ -d ${DP_BACKUP_DIR} ]; then
-    rm -rf ${DP_BACKUP_DIR}
-fi
-mkdir -p ${DP_BACKUP_DIR}
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 START_TIME=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
 xtrabackup --compress=zstd --backup --safe-slave-backup --slave-info --stream=xbstream \
-    --host=${DP_DB_HOST} --user=${DP_DB_USER} --port=${DP_DB_PORT} --password=${DP_DB_PASSWORD} --datadir=${DATA_DIR} >${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.xbstream
+    --host=${DP_DB_HOST} --port=${DP_DB_PORT} \
+    --user=${DP_DB_USER} --password=${DP_DB_PASSWORD} \
+    --datadir=${DATA_DIR} | datasafed push - "/${DP_BACKUP_NAME}.xbstream"
 STOP_TIME=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
-TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR} | awk '{print $1}')
-echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" >${DP_BACKUP_DIR}/backup.info
+TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" > "${DP_BACKUP_INFO_FILE}"
diff --git a/deploy/apecloud-mysql/dataprotection/restore.sh b/deploy/apecloud-mysql/dataprotection/restore.sh
index 89f6b689d74..7a00e6cf93f 100644
--- a/deploy/apecloud-mysql/dataprotection/restore.sh
+++ b/deploy/apecloud-mysql/dataprotection/restore.sh
@@ -1,9 +1,12 @@
 #!/bin/bash
 set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 mkdir -p ${DATA_DIR}
 TMP_DIR=${DATA_MOUNT_DIR}/temp
 mkdir -p ${TMP_DIR} && cd ${TMP_DIR}
-xbstream -x <${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.xbstream
+datasafed pull "${DP_BACKUP_NAME}.xbstream" - | xbstream -x
 xtrabackup --decompress --remove-original --target-dir=${TMP_DIR}
 xtrabackup --prepare --target-dir=${TMP_DIR}
 xtrabackup --move-back --target-dir=${TMP_DIR} --datadir=${DATA_DIR}/ --log-bin=${LOG_BIN}
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_backuprepos.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_backuprepos.yaml
index 00fcc2ab7a4..43188e28b3d 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_backuprepos.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_backuprepos.yaml
@@ -24,6 +24,9 @@ spec:
     - jsonPath: .spec.storageProviderRef
       name: STORAGEPROVIDER
       type: string
+    - jsonPath: .spec.accessMethod
+      name: ACCESSMETHOD
+      type: string
     - jsonPath: .status.isDefault
       name: DEFAULT
       type: boolean
diff --git a/deploy/helm/templates/addons/csi-s3-addon.yaml b/deploy/helm/templates/addons/csi-s3-addon.yaml
index 834424433a0..74fd9c293ba 100644
--- a/deploy/helm/templates/addons/csi-s3-addon.yaml
+++ b/deploy/helm/templates/addons/csi-s3-addon.yaml
@@ -51,8 +51,10 @@ spec:
     {{- $autoInstall := (get ( get ( .Values | toYaml | fromYaml ) "csi-s3" ) "enabled") }}
     {{- /* auto install csi-s3 if it's required by backup repos */ -}}
     {{- if .Values.backupRepo.create }}
-      {{- if eq .Values.backupRepo.storageProvider "s3" "oss" "minio" "obs" "cos" "gcs" }}
-        {{- $autoInstall = true }}
+      {{- if eq .Values.backupRepo.accessMethod "Mount" }}
+        {{- if eq .Values.backupRepo.storageProvider "s3" "oss" "minio" "obs" "cos" "gcs" }}
+          {{- $autoInstall = true }}
+        {{- end }}
       {{- end }}
     {{- end }}
     autoInstall: {{ $autoInstall }}
diff --git a/deploy/helm/templates/dataprotection.yaml b/deploy/helm/templates/dataprotection.yaml
index 15e6b3a0fb3..b02a0d9ba8a 100644
--- a/deploy/helm/templates/dataprotection.yaml
+++ b/deploy/helm/templates/dataprotection.yaml
@@ -87,6 +87,8 @@ spec:
             - name: ENABLE_WEBHOOKS
               value: "true"
             {{- end }}
+            - name: DATASAFED_IMAGE
+              value: "{{ .Values.dataProtection.image.registry | default "docker.io" }}/{{ .Values.dataProtection.image.datasafed.repository }}:{{ .Values.dataProtection.image.datasafed.tag | default "latest" }}"
           {{- with .Values.securityContext }}
           securityContext:
             {{- toYaml . | nindent 12 }}
diff --git a/deploy/helm/templates/storageprovider/cos.yaml b/deploy/helm/templates/storageprovider/cos.yaml
index ec39ede7d5a..07cb399f72b 100644
--- a/deploy/helm/templates/storageprovider/cos.yaml
+++ b/deploy/helm/templates/storageprovider/cos.yaml
@@ -42,6 +42,7 @@ spec:
     secret_access_key = {{ `{{ index .Parameters "secretAccessKey" }}` }}
     endpoint = {{ `{{ printf "cos.%s.myqcloud.com" .Parameters.region }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/templates/storageprovider/gcs.yaml b/deploy/helm/templates/storageprovider/gcs.yaml
index 0d8ae56f63b..2e38d9c8c68 100644
--- a/deploy/helm/templates/storageprovider/gcs.yaml
+++ b/deploy/helm/templates/storageprovider/gcs.yaml
@@ -45,6 +45,7 @@ spec:
     {{ `{{- end }}` }}
     endpoint = {{ `{{ $endpoint }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/templates/storageprovider/minio.yaml b/deploy/helm/templates/storageprovider/minio.yaml
index 62ff8b65489..fde1c53f0f0 100644
--- a/deploy/helm/templates/storageprovider/minio.yaml
+++ b/deploy/helm/templates/storageprovider/minio.yaml
@@ -36,6 +36,7 @@ spec:
     secret_access_key = {{ `{{ index .Parameters "secretAccessKey" }}` }}
     endpoint = {{ `{{ index .Parameters "endpoint" }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/templates/storageprovider/obs.yaml b/deploy/helm/templates/storageprovider/obs.yaml
index a61ea97b19a..237e534269c 100644
--- a/deploy/helm/templates/storageprovider/obs.yaml
+++ b/deploy/helm/templates/storageprovider/obs.yaml
@@ -43,6 +43,7 @@ spec:
     region = {{ `{{ index .Parameters "region" }}` }}
     endpoint = {{ `{{ printf "obs.%s.myhuaweicloud.com" .Parameters.region }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/templates/storageprovider/oss.yaml b/deploy/helm/templates/storageprovider/oss.yaml
index 5ff3b780e95..3d6ca910daa 100644
--- a/deploy/helm/templates/storageprovider/oss.yaml
+++ b/deploy/helm/templates/storageprovider/oss.yaml
@@ -41,6 +41,7 @@ spec:
     secret_access_key = {{ `{{ index .Parameters "secretAccessKey" }}` }}
     endpoint = {{ `{{- printf "oss-%s.aliyuncs.com" .Parameters.region) }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/templates/storageprovider/s3.yaml b/deploy/helm/templates/storageprovider/s3.yaml
index 8bc6730d1b6..ef5fd6d4c9d 100644
--- a/deploy/helm/templates/storageprovider/s3.yaml
+++ b/deploy/helm/templates/storageprovider/s3.yaml
@@ -46,6 +46,7 @@ spec:
     region = {{ `{{ index .Parameters "region" }}` }}
     endpoint = {{ `{{ index .Parameters "endpoint" }}` }}
     root = {{ `{{ index .Parameters "bucket" }}` }}
+    chunk_size = 50Mi
 
   parametersSchema:
     openAPIV3Schema:
diff --git a/deploy/helm/values.yaml b/deploy/helm/values.yaml
index 2faf409e0ed..aed3dda45cd 100644
--- a/deploy/helm/values.yaml
+++ b/deploy/helm/values.yaml
@@ -322,11 +322,15 @@ dataProtection:
     # Overrides the image tag whose default is the chart appVersion.
     tag: ""
     imagePullSecrets: []
+    datasafed:
+      repository: apecloud/datasafed
+      tag: "0.0.3"
 
 ## BackupRepo settings
 ##
 ## @param backupRepo.create - creates a backup repo during installation
 ## @param backupRepo.default - set the created repo as the default
+## @param backupRepo.accessMethod - the access method for the backup repo, options: [Mount, Tool]
 ## @param backupRepo.storageProvider - the storage provider used by the repo, options: [s3, oss, minio]
 ## @param backupRepo.pvReclaimPolicy - the PV reclaim policy, options: [Retain, Delete]
 ## @param backupRepo.volumeCapacity - the capacity for creating PVC
@@ -335,6 +339,7 @@ dataProtection:
 backupRepo:
   create: false
   default: true
+  accessMethod: Tool
   storageProvider: ""
   pvReclaimPolicy: "Retain"
   volumeCapacity: ""
diff --git a/deploy/mongodb/dataprotection/backup-info-collector.sh b/deploy/mongodb/dataprotection/backup-info-collector.sh
index e3d2b63d4d3..7ef9cb063a7 100644
--- a/deploy/mongodb/dataprotection/backup-info-collector.sh
+++ b/deploy/mongodb/dataprotection/backup-info-collector.sh
@@ -6,11 +6,13 @@ function get_current_time() {
 }
 
 function stat_and_save_backup_info() {
+  export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+  export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
   START_TIME=$1
   STOP_TIME=$2
   if [ -z $STOP_TIME ]; then
     STOP_TIME=`get_current_time`
   fi
-  TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-  echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" > ${DP_BACKUP_DIR}/backup.info
+  TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+  echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" > "${DP_BACKUP_INFO_FILE}"
 }
\ No newline at end of file
diff --git a/deploy/mongodb/dataprotection/datafile-backup.sh b/deploy/mongodb/dataprotection/datafile-backup.sh
index d5eee06e666..85a685de166 100644
--- a/deploy/mongodb/dataprotection/datafile-backup.sh
+++ b/deploy/mongodb/dataprotection/datafile-backup.sh
@@ -1,10 +1,11 @@
-if [ -d ${DP_BACKUP_DIR} ]; then
-  rm -rf ${DP_BACKUP_DIR}
-fi
-mkdir -p ${DP_BACKUP_DIR} && cd ${DATA_DIR}
+set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
+cd ${DATA_DIR}
 START_TIME=`get_current_time`
 # TODO: flush data and locked write, otherwise data maybe inconsistent
-tar -czvf ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.tar.gz ./
+tar -czvf - ./ | datasafed push - "${DP_BACKUP_NAME}.tar.gz"
 rm -rf mongodb.backup
 # stat and save the backup information
 stat_and_save_backup_info $START_TIME
\ No newline at end of file
diff --git a/deploy/mongodb/dataprotection/datafile-restore.sh b/deploy/mongodb/dataprotection/datafile-restore.sh
index a527c4b9528..b07ae478126 100644
--- a/deploy/mongodb/dataprotection/datafile-restore.sh
+++ b/deploy/mongodb/dataprotection/datafile-restore.sh
@@ -1,4 +1,7 @@
 set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 mkdir -p ${DATA_DIR}
 res=`ls -A ${DATA_DIR}`
 data_protection_file=${DATA_DIR}/.kb-data-protection
@@ -8,5 +11,5 @@ if [ ! -z "${res}" ] && [ ! -f ${data_protection_file} ]; then
 fi
 cd ${DATA_DIR} && touch mongodb.backup
 touch ${data_protection_file}
-tar -xvf ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.tar.gz -C ${DATA_DIR}
+datasafed pull "${DP_BACKUP_NAME}.tar.gz" - | tar -xzvf - -C ${DATA_DIR}
 rm -rf ${data_protection_file} && sync
\ No newline at end of file
diff --git a/deploy/mongodb/dataprotection/mongodump-backup.sh b/deploy/mongodb/dataprotection/mongodump-backup.sh
index 0bae45552c2..d600d1e717b 100644
--- a/deploy/mongodb/dataprotection/mongodump-backup.sh
+++ b/deploy/mongodb/dataprotection/mongodump-backup.sh
@@ -1,12 +1,12 @@
-if [ -d ${DP_BACKUP_DIR} ]; then
-  rm -rf ${DP_BACKUP_DIR}
-fi
-mkdir -p ${DP_BACKUP_DIR}
+set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 
 # TODO: support endpoint env for sharding cluster.
 mongo_uri="mongodb://${DP_DB_HOST}:${DP_DB_PORT}"
 START_TIME=`get_current_time`
-mongodump --uri ${mongo_uri} -u ${DP_DB_USER} -p ${DP_DB_PASSWORD} --authenticationDatabase admin --out ${DP_BACKUP_DIR}
+mongodump --uri "${mongo_uri}" -u ${DP_DB_USER} -p ${DP_DB_PASSWORD} --authenticationDatabase admin --archive --gzip | datasafed push - "${DP_BACKUP_NAME}.archive"
 
 # stat and save the backup information
 stat_and_save_backup_info $START_TIME
\ No newline at end of file
diff --git a/deploy/mongodb/dataprotection/mongodump-restore.sh b/deploy/mongodb/dataprotection/mongodump-restore.sh
index 66ebfe5b4f2..238ba369499 100644
--- a/deploy/mongodb/dataprotection/mongodump-restore.sh
+++ b/deploy/mongodb/dataprotection/mongodump-restore.sh
@@ -1,6 +1,7 @@
+set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
+
 mongo_uri="mongodb://${DP_DB_HOST}:${DP_DB_PORT}"
-for dir_name in $(ls ${DP_BACKUP_DIR} -l | grep ^d | awk '{print $9}'); do
-  database_dir=${DP_BACKUP_DIR}/$dir_name
-  echo "INFO: restoring from ${database_dir}"
-  mongorestore --uri ${mongo_uri} -u ${MONGODB_ROOT_USER} -p ${MONGODB_ROOT_PASSWORD} -d $dir_name --authenticationDatabase admin ${database_dir}
-done
\ No newline at end of file
+datasafed pull "${DP_BACKUP_NAME}.archive" - | mongorestore --archive --gzip --uri "${mongo_uri}" -u ${MONGODB_ROOT_USER} -p ${MONGODB_ROOT_PASSWORD} --authenticationDatabase admin
diff --git a/deploy/mongodb/dataprotection/pitr-backup.sh b/deploy/mongodb/dataprotection/pitr-backup.sh
deleted file mode 100644
index eb9cfd80e2d..00000000000
--- a/deploy/mongodb/dataprotection/pitr-backup.sh
+++ /dev/null
@@ -1,83 +0,0 @@
-#!/bin/bash
-mkdir -p ${DP_BACKUP_DIR} && cd ${DP_BACKUP_DIR}
-# retention 8 days by default
-retention_minute=""
-if [ ! -z ${LOGFILE_TTL_SECOND} ];then
-  retention_minute=$((${LOGFILE_TTL_SECOND}/60))
-fi
-export MONGODB_URI="mongodb://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:27017/?authSource=admin"
-export WALG_FILE_PREFIX=${DP_BACKUP_DIR}
-export OPLOG_ARCHIVE_TIMEOUT_INTERVAL=${ARCHIVE_INTERVAL}
-export OPLOG_ARCHIVE_AFTER_SIZE=${ARCHIVE_AFTER_SIZE}
-retryTimes=0
-purgeCounter=0
-wal_g_pid=0
-
-do_oplog_push(){
-  echo "start to archive oplog..."
-  echo "wal-g oplog-push > /tmp/wal-g-oplog.log"
-  wal-g oplog-push >/tmp/wal-g-oplog.log 2>&1 &
-  wal_g_pid=$!
-  sleep 1
-  cat /tmp/wal-g-oplog.log
-}
-
-check_oplog_push_process(){
-  # check wal-g oplog-push process
-  ps -p $wal_g_pid >/dev/null
-  if [ $? -ne 0 ]; then
-    echo 'ERROR: the process "wal-g oplog-push" does not exist!'
-    errorLog=$(cat /tmp/wal-g-oplog.log)
-    echo $errorLog && exit 1
-  fi
-  # check role of the connected mongodb
-  CLIENT=`which mongosh>/dev/null&&echo mongosh||echo mongo`
-  isPrimary=$(${CLIENT} -u ${DB_USER} -p ${DB_PASSWORD} --port 27017 --host ${DB_HOST} --authenticationDatabase admin  --eval 'db.isMaster().ismaster' --quiet)
-  if [ "${isPrimary}" != "true" ]; then
-    echo "isPrimary: ${isPrimary}"
-    retryTimes=$(($retryTimes+1))
-  else
-    retryTimes=0
-  fi
-  if [ $retryTimes -ge 3 ]; then
-     echo "ERROR: the current mongo instance is not a primary node, 3 attempts have been made!" && kill $wal_g_pid
-  fi
-}
-
-save_backup_status() {
-   TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-   OLDEST_FILE=$(ls -t ${DP_BACKUP_DIR}/oplog_005 | tail -n 1) && OLDEST_FILE=${OLDEST_FILE#*_} && LOG_START_TIME=${OLDEST_FILE%%.*}
-   LATEST_FILE=$(ls -t ${DP_BACKUP_DIR}/oplog_005 | head -n 1) && LATEST_FILE=${LATEST_FILE##*_} && LOG_STOP_TIME=${LATEST_FILE%%.*}
-   if [ ! -z $LOG_START_TIME ]; then
-       START_TIME=$(date -d "@${LOG_START_TIME}" -u '+%Y-%m-%dT%H:%M:%SZ')
-       STOP_TIME=$(date -d "@${LOG_STOP_TIME}" -u '+%Y-%m-%dT%H:%M:%SZ')
-       echo "{\"totalSize\":\"$TOTAL_SIZE\",\"manifests\":{\"backupLog\":{\"startTime\":\"${START_TIME}\",\"stopTime\":\"${STOP_TIME}\"},\"backupTool\":{\"uploadTotalSize\":\"${TOTAL_SIZE}\"}}}" > ${DP_BACKUP_DIR}/backup.info
-   fi
-}
-# purge the expired files
-purge_expired_files() {
-  if [ ! -z ${LOGFILE_TTL_SECOND} ];then
-    purgeCounter=$((purgeCounter+3))
-    if [ $purgeCounter -ge 60 ]; then
-       purgeCounter=0
-       fileCount=$(find ${DP_BACKUP_DIR}/oplog_005 -mmin +${retention_minute} -name "*.lz4" | wc -l)
-       find ${DP_BACKUP_DIR}/oplog_005 -mmin +${retention_minute} -name "*.lz4" -exec rm -rf {} \;
-       if [ ${fileCount} -gt 0 ]; then
-          echo "clean up expired oplog file successfully, file count: ${fileCount}"
-       fi
-    fi
-  fi
-}
-# create oplog push process
-do_oplog_push
-# trap term signal
-trap "echo 'Terminating...' && kill $wal_g_pid" TERM
-while true; do
-  check_oplog_push_process
-  sleep 1
-  if [ -d ${DP_BACKUP_DIR}/oplog_005 ];then
-    save_backup_status
-    # purge the expired oplog
-    purge_expired_files
-  fi
-done
\ No newline at end of file
diff --git a/deploy/mongodb/scripts/replicaset-restore.tpl b/deploy/mongodb/scripts/replicaset-restore.tpl
deleted file mode 100644
index fa758a92ba7..00000000000
--- a/deploy/mongodb/scripts/replicaset-restore.tpl
+++ /dev/null
@@ -1,29 +0,0 @@
-#!/bin/sh
-
-set -e
-PORT=27017
-MONGODB_ROOT=/data/mongodb
-mkdir -p $MONGODB_ROOT/db
-mkdir -p $MONGODB_ROOT/logs
-mkdir -p $MONGODB_ROOT/tmp
-
-res=`ls -A ${DATA_DIR}`
-if [ ! -z ${res} ]; then
-  echo "${DATA_DIR} is not empty! Please make sure that the directory is empty before restoring the backup."
-  exit 1
-fi
-tar -xvf ${BACKUP_DIR}/${BACKUP_NAME}.tar.gz -C ${DATA_DIR}/../
-mv ${DATA_DIR}/../${BACKUP_NAME}/* ${DATA_DIR}
-RPL_SET_NAME=$(echo $KB_POD_NAME | grep -o ".*-");
-RPL_SET_NAME=${RPL_SET_NAME%-};
-MODE=$1
-mongod $MODE --bind_ip_all --port $PORT --dbpath $MONGODB_ROOT/db --directoryperdb --logpath $MONGODB_ROOT/logs/mongodb.log  --logappend --pidfilepath $MONGODB_ROOT/tmp/mongodb.pid&
-export CLIENT=`which mongosh>/dev/null&&echo mongosh||echo mongo`
-until $CLIENT --quiet --port $PORT --host $host --eval "print('peer is ready')"; do sleep 1; done
-PID=`cat $MONGODB_ROOT/tmp/mongodb.pid`
-
-$CLIENT --quiet --port $PORT local --eval "db.system.replset.deleteOne({})"
-$CLIENT --quiet --port $PORT local --eval "db.system.replset.find()"
-$CLIENT --quiet --port $PORT admin --eval 'db.dropUser("root", {w: "majority", wtimeout: 4000})' || true
-kill $PID
-wait $PID
diff --git a/deploy/oracle-mysql/templates/actionset-xtrabackup.yaml b/deploy/oracle-mysql/templates/actionset-xtrabackup.yaml
index 243bb87e2e2..0862b875dee 100644
--- a/deploy/oracle-mysql/templates/actionset-xtrabackup.yaml
+++ b/deploy/oracle-mysql/templates/actionset-xtrabackup.yaml
@@ -20,12 +20,14 @@ spec:
       - bash
       - -c
       - |
-        set -e;
-        mkdir -p ${DP_BACKUP_DIR};
+        set -e
+        set -o pipefail
+        export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+        export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
         xtrabackup --backup --safe-slave-backup --slave-info --stream=xbstream \
-        --host=${DP_DB_HOST} --user=${DP_DB_USER} --password=${DP_DB_PASSWORD} --datadir=${DATA_DIR} > ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.xbstream
-        TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-        echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > ${DP_BACKUP_DIR}/backup.info
+        --host=${DP_DB_HOST} --user=${DP_DB_USER} --password=${DP_DB_PASSWORD} --datadir=${DATA_DIR} | datasafed push - "/${DP_BACKUP_NAME}.xbstream"
+        TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+        echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > "${DP_BACKUP_INFO_FILE}"
       syncProgress:
         enabled: true
         intervalSeconds: 5
@@ -36,11 +38,14 @@ spec:
       - bash
       - -c
       - |
-        set -e;
+        set -e
+        set -o pipefail
+        export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+        export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
         mkdir -p ${DATA_DIR}
         TMP_DIR=/data/mysql/temp
         mkdir -p ${TMP_DIR} && cd ${TMP_DIR}
-        xbstream -x < ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.xbstream
+        datasafed pull "${DP_BACKUP_NAME}.xbstream" - | xbstream -x
         xtrabackup --decompress --remove-original --target-dir=${TMP_DIR}
         xtrabackup --prepare --target-dir=${TMP_DIR}
         xtrabackup --move-back --target-dir=${TMP_DIR} --datadir=${DATA_DIR}/
diff --git a/deploy/postgresql/dataprotection/backup-info-collector.sh b/deploy/postgresql/dataprotection/backup-info-collector.sh
index 2324b0b2a1d..77886dd7823 100644
--- a/deploy/postgresql/dataprotection/backup-info-collector.sh
+++ b/deploy/postgresql/dataprotection/backup-info-collector.sh
@@ -4,6 +4,8 @@ function get_current_time() {
 }
 
 function stat_and_save_backup_info() {
+  export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+  export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
   START_TIME=$1
   STOP_TIME=$2
   if [ -z $STOP_TIME ]; then
@@ -11,6 +13,6 @@ function stat_and_save_backup_info() {
   fi
   START_TIME=$(date -d "${START_TIME}" -u '+%Y-%m-%dT%H:%M:%SZ')
   STOP_TIME=$(date -d "${STOP_TIME}" -u '+%Y-%m-%dT%H:%M:%SZ')
-  TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-  echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" > ${DP_BACKUP_DIR}/backup.info
+  TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+  echo "{\"totalSize\":\"$TOTAL_SIZE\",\"timeRange\":{\"start\":\"${START_TIME}\",\"end\":\"${STOP_TIME}\"}}" > "${DP_BACKUP_INFO_FILE}"
 }
\ No newline at end of file
diff --git a/deploy/postgresql/dataprotection/pg-basebackup-backup.sh b/deploy/postgresql/dataprotection/pg-basebackup-backup.sh
index 0fb6296841a..9fb0cd540eb 100644
--- a/deploy/postgresql/dataprotection/pg-basebackup-backup.sh
+++ b/deploy/postgresql/dataprotection/pg-basebackup-backup.sh
@@ -1,12 +1,11 @@
-set -e;
-if [ -d ${DP_BACKUP_DIR} ]; then
-  rm -rf ${DP_BACKUP_DIR}
-fi
-mkdir -p ${DP_BACKUP_DIR};
+set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 export PGPASSWORD=${DP_DB_PASSWORD}
 
 START_TIME=`get_current_time`
-echo ${DP_DB_PASSWORD} | pg_basebackup -Ft -Pv -c fast -Xs -Z${COMPRESS_LEVEL} -D ${DP_BACKUP_DIR} -h ${DP_DB_HOST} -U ${DP_DB_USER} -W;
+echo ${DP_DB_PASSWORD} | pg_basebackup -Ft -Pv -c fast -Xf -D - -h ${DP_DB_HOST} -U ${DP_DB_USER} -W | gzip | datasafed push - "/${DP_BACKUP_NAME}.tar.gz"
 
 # stat and save the backup information
 stat_and_save_backup_info $START_TIME
\ No newline at end of file
diff --git a/deploy/postgresql/dataprotection/pg-basebackup-restore.sh b/deploy/postgresql/dataprotection/pg-basebackup-restore.sh
index 1be2f6ec70a..d15a1a68218 100644
--- a/deploy/postgresql/dataprotection/pg-basebackup-restore.sh
+++ b/deploy/postgresql/dataprotection/pg-basebackup-restore.sh
@@ -1,15 +1,7 @@
-set -e;
-cd ${DP_BACKUP_DIR};
+set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 mkdir -p ${DATA_DIR};
-# compatible with gzip compression
-if [ -f base.tar.gz ];then
-  tar -xvf base.tar.gz -C ${DATA_DIR}/;
-else
-  tar -xvf base.tar -C ${DATA_DIR}/;
-fi
-if [ -f pg_wal.tar.gz ];then
-  tar -xvf pg_wal.tar.gz -C ${DATA_DIR}/pg_wal/;
-else
-  tar -xvf pg_wal.tar -C ${DATA_DIR}/pg_wal/;
-fi
+datasafed pull "${DP_BACKUP_NAME}.tar.gz" - | gunzip | tar -xvf - -C "${DATA_DIR}/"
 echo "done!";
\ No newline at end of file
diff --git a/deploy/postgresql/templates/actionset-pgbasebackup.yaml b/deploy/postgresql/templates/actionset-pgbasebackup.yaml
index 94e8904d0da..dc706bbc3b7 100644
--- a/deploy/postgresql/templates/actionset-pgbasebackup.yaml
+++ b/deploy/postgresql/templates/actionset-pgbasebackup.yaml
@@ -10,8 +10,6 @@ spec:
   env:
     - name: DATA_DIR
       value: {{ .Values.dataMountPath }}/pgroot/data
-    - name: COMPRESS_LEVEL
-      value: "0"
   backup:
     preBackup: []
     postBackup: []
diff --git a/deploy/qdrant/scripts/qdrant-backup.sh b/deploy/qdrant/scripts/qdrant-backup.sh
index a8e5da4be97..e26b0906d5f 100644
--- a/deploy/qdrant/scripts/qdrant-backup.sh
+++ b/deploy/qdrant/scripts/qdrant-backup.sh
@@ -1,7 +1,10 @@
 #!/usr/bin/env bash
 
 set -e
-mkdir -p ${DP_BACKUP_DIR}
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
+
 endpoint=http://${DP_DB_HOST}:6333
 
 snapshot=`curl -XPOST ${endpoint}/snapshots`
@@ -11,10 +14,10 @@ if [ "${status}" != "ok" ] && [ "${status}" != "\"ok\"" ]; then
     exit 1
 fi
 
-name=`echo ${snapshot} | jq '.result.name'`
-curl ${endpoint}/snapshots/${name} --output ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.snapshot
+name=`echo ${snapshot} | jq -r '.result.name'`
+curl -v --fail-with-body ${endpoint}/snapshots/${name} | datasafed push - "/${DP_BACKUP_NAME}.snapshot"
 
 curl -XDELETE ${endpoint}/snapshots/${name}
 
-TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > ${DP_BACKUP_DIR}/backup.info
\ No newline at end of file
+TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > "${DP_BACKUP_INFO_FILE}"
\ No newline at end of file
diff --git a/deploy/qdrant/scripts/qdrant-restore.sh b/deploy/qdrant/scripts/qdrant-restore.sh
index fc0d3007cee..a913679febe 100644
--- a/deploy/qdrant/scripts/qdrant-restore.sh
+++ b/deploy/qdrant/scripts/qdrant-restore.sh
@@ -1,6 +1,9 @@
 #!/usr/bin/env bash
 
 set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 mkdir -p ${DATA_DIR}
 res=`ls -A ${DATA_DIR}`
 if [ ! -z "${res}" ]; then
@@ -8,8 +11,14 @@ if [ ! -z "${res}" ]; then
   exit 1
 fi
 
+# download snapshot file
+SNAPSHOT_DIR="${DATA_DIR}/_dp_snapshots"
+SNAPSHOT_FILE="${DP_BACKUP_NAME}.snapshot"
+mkdir -p "${SNAPSHOT_DIR}"
+datasafed pull "${SNAPSHOT_FILE}" "${SNAPSHOT_DIR}/${SNAPSHOT_FILE}"
+
 # start qdrant restore process
-qdrant --storage-snapshot ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}  --config-path /qdrant/config/config.yaml  --force-snapshot --uri http://localhost:6333 &
+qdrant --storage-snapshot "${SNAPSHOT_DIR}/${SNAPSHOT_FILE}"  --config-path /qdrant/config/config.yaml  --force-snapshot --uri http://localhost:6333 &
 
 # wait until restore finished
 until curl http://localhost:6333/cluster; do sleep 1; done
@@ -18,3 +27,6 @@ until curl http://localhost:6333/cluster; do sleep 1; done
 pid=`pidof qdrant`
 kill -s INT ${pid}
 wait ${pid}
+
+# delete snapshot file
+rm -rf "${SNAPSHOT_DIR}"
diff --git a/deploy/redis/dataprotection/backup.sh b/deploy/redis/dataprotection/backup.sh
index a60057b4a7e..462e9bb5e54 100644
--- a/deploy/redis/dataprotection/backup.sh
+++ b/deploy/redis/dataprotection/backup.sh
@@ -1,4 +1,7 @@
 set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 connect_url="redis-cli -h ${DP_DB_HOST} -p ${DP_DB_PORT} -a ${DP_DB_PASSWORD}"
 last_save=$(${connect_url} LASTSAVE)
 echo "INFO: start BGSAVE"
@@ -12,8 +15,8 @@ while true; do
   sleep 1
 done
 echo "INFO: start to save data file..."
-mkdir -p ${DP_BACKUP_DIR} && cd ${DATA_DIR}
-tar -czvf ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.tar.gz ./
+cd ${DATA_DIR}
+tar -czvf - ./ | datasafed push - "${DP_BACKUP_NAME}.tar.gz"
 echo "INFO: save data file successfully"
-TOTAL_SIZE=$(du -shx ${DP_BACKUP_DIR}|awk '{print $1}')
-echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > ${DP_BACKUP_DIR}/backup.info && sync
\ No newline at end of file
+TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+echo "{\"totalSize\":\"$TOTAL_SIZE\"}" > "${DP_BACKUP_INFO_FILE}" && sync
diff --git a/deploy/redis/dataprotection/restore.sh b/deploy/redis/dataprotection/restore.sh
index c1e72b0f2ed..17ade60ed69 100644
--- a/deploy/redis/dataprotection/restore.sh
+++ b/deploy/redis/dataprotection/restore.sh
@@ -1,4 +1,7 @@
 set -e
+set -o pipefail
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH="$DP_BACKUP_BASE_PATH"
 mkdir -p ${DATA_DIR}
 res=`find ${DATA_DIR} -type f`
 data_protection_file=${DATA_DIR}/.kb-data-protection
@@ -8,5 +11,5 @@ if [ ! -z "${res}" ] && [ ! -f ${data_protection_file} ]; then
 fi
 # touch placeholder file
 touch ${data_protection_file}
-tar -xvf ${DP_BACKUP_DIR}/${DP_BACKUP_NAME}.tar.gz -C ${DATA_DIR}
+datasafed pull "${DP_BACKUP_NAME}.tar.gz" - | tar -xzvf - -C ${DATA_DIR}
 rm -rf ${data_protection_file} && sync
\ No newline at end of file
diff --git a/deploy/redis/templates/backupactionset.yaml b/deploy/redis/templates/backupactionset.yaml
index f406fa92461..45348e0048d 100644
--- a/deploy/redis/templates/backupactionset.yaml
+++ b/deploy/redis/templates/backupactionset.yaml
@@ -30,7 +30,7 @@ spec:
     prepareData:
       image: {{ include "redis.image" . }}
       command:
-      - sh
+      - bash
       - -c
       - |
         {{- .Files.Get "dataprotection/restore.sh" | nindent 8 }}
diff --git a/docs/user_docs/cli/cli.md b/docs/user_docs/cli/cli.md
index 9d3e281d01a..e8fc0abaa6f 100644
--- a/docs/user_docs/cli/cli.md
+++ b/docs/user_docs/cli/cli.md
@@ -30,8 +30,8 @@ Manage alert receiver, include add, list and delete receiver.
 BackupRepo command.
 
 * [kbcli backuprepo create](kbcli_backuprepo_create.md)	 - Create a backup repo
-* [kbcli backuprepo describe](kbcli_backuprepo_describe.md)	 - Describe a backuprepo.
-* [kbcli backuprepo list](kbcli_backuprepo_list.md)	 - List BackupRepo.
+* [kbcli backuprepo describe](kbcli_backuprepo_describe.md)	 - Describe a backup repository.
+* [kbcli backuprepo list](kbcli_backuprepo_list.md)	 - List Backup Repositories.
 
 
 ## [bench](kbcli_bench.md)
diff --git a/docs/user_docs/cli/kbcli_backuprepo.md b/docs/user_docs/cli/kbcli_backuprepo.md
index 79e85152c27..4c33bea0dcb 100644
--- a/docs/user_docs/cli/kbcli_backuprepo.md
+++ b/docs/user_docs/cli/kbcli_backuprepo.md
@@ -38,8 +38,8 @@ BackupRepo command.
 
 
 * [kbcli backuprepo create](kbcli_backuprepo_create.md)	 - Create a backup repo
-* [kbcli backuprepo describe](kbcli_backuprepo_describe.md)	 - Describe a backuprepo.
-* [kbcli backuprepo list](kbcli_backuprepo_list.md)	 - List BackupRepo.
+* [kbcli backuprepo describe](kbcli_backuprepo_describe.md)	 - Describe a backup repository.
+* [kbcli backuprepo list](kbcli_backuprepo_list.md)	 - List Backup Repositories.
 
 #### Go Back to [CLI Overview](cli.md) Homepage.
 
diff --git a/docs/user_docs/cli/kbcli_backuprepo_create.md b/docs/user_docs/cli/kbcli_backuprepo_create.md
index f052d417d2e..f966aa512ff 100644
--- a/docs/user_docs/cli/kbcli_backuprepo_create.md
+++ b/docs/user_docs/cli/kbcli_backuprepo_create.md
@@ -32,6 +32,7 @@ kbcli backuprepo create [NAME] [flags]
 ### Options
 
 ```
+      --access-method string       Specify the access method for the backup repository, "Tool" is preferred if not specified. options: ["Mount" "Tool"]
       --default                    Specify whether to set the created backup repo as default
   -h, --help                       help for create
       --provider string            Specify storage provider
diff --git a/docs/user_docs/cli/kbcli_backuprepo_describe.md b/docs/user_docs/cli/kbcli_backuprepo_describe.md
index 986b62b6e03..5a176b0527a 100644
--- a/docs/user_docs/cli/kbcli_backuprepo_describe.md
+++ b/docs/user_docs/cli/kbcli_backuprepo_describe.md
@@ -2,7 +2,7 @@
 title: kbcli backuprepo describe
 ---
 
-Describe a backuprepo.
+Describe a backup repository.
 
 ```
 kbcli backuprepo describe [flags]
diff --git a/docs/user_docs/cli/kbcli_backuprepo_list.md b/docs/user_docs/cli/kbcli_backuprepo_list.md
index eecde3de334..575f9c17b26 100644
--- a/docs/user_docs/cli/kbcli_backuprepo_list.md
+++ b/docs/user_docs/cli/kbcli_backuprepo_list.md
@@ -2,7 +2,7 @@
 title: kbcli backuprepo list
 ---
 
-List BackupRepo.
+List Backup Repositories.
 
 ```
 kbcli backuprepo list [flags]
@@ -11,7 +11,7 @@ kbcli backuprepo list [flags]
 ### Examples
 
 ```
-  # List all backuprepos
+  # List all backup repositories
   kbcli backuprepo list
 ```
 
diff --git a/internal/cli/cmd/backuprepo/create.go b/internal/cli/cmd/backuprepo/create.go
index fa131bd465d..b4420edffc1 100644
--- a/internal/cli/cmd/backuprepo/create.go
+++ b/internal/cli/cmd/backuprepo/create.go
@@ -61,7 +61,14 @@ const (
 )
 
 var (
-	allowedPVReclaimPolicies = []string{"Retain", "Delete"}
+	allowedAccessMethods = []string{
+		string(dpv1alpha1.AccessMethodMount),
+		string(dpv1alpha1.AccessMethodTool),
+	}
+	allowedPVReclaimPolicies = []string{
+		string(corev1.PersistentVolumeReclaimRetain),
+		string(corev1.PersistentVolumeReclaimDelete),
+	}
 )
 
 type createOptions struct {
@@ -70,6 +77,7 @@ type createOptions struct {
 	client  kubernetes.Interface
 	factory cmdutil.Factory
 
+	accessMethod    string
 	storageProvider string
 	providerObject  *storagev1alpha1.StorageProvider
 	isDefault       bool
@@ -124,6 +132,8 @@ func newCreateCommand(o *createOptions, f cmdutil.Factory, streams genericioopti
 		},
 		DisableFlagParsing: true,
 	}
+	cmd.Flags().StringVar(&o.accessMethod, "access-method", "",
+		fmt.Sprintf("Specify the access method for the backup repository, \"Tool\" is preferred if not specified. options: %q", allowedAccessMethods))
 	cmd.Flags().StringVar(&o.storageProvider, providerFlagName, "", "Specify storage provider")
 	util.CheckErr(cmd.MarkFlagRequired(providerFlagName))
 	cmd.Flags().BoolVar(&o.isDefault, "default", false, "Specify whether to set the created backup repo as default")
@@ -164,12 +174,23 @@ func flagsToValues(fs *pflag.FlagSet) map[string]string {
 
 func (o *createOptions) parseProviderFlags(cmd *cobra.Command, args []string, f cmdutil.Factory) error {
 	// Since we disabled the flag parsing of the cmd, we need to parse it from args
+	help := false
 	tmpFlags := pflag.NewFlagSet("tmp", pflag.ContinueOnError)
 	tmpFlags.StringVar(&o.storageProvider, providerFlagName, "", "")
-	tmpFlags.BoolP("help", "h", false, "") // eat --help and -h
+	tmpFlags.BoolVarP(&help, "help", "h", false, "") // eat --help and -h
 	tmpFlags.ParseErrorsWhitelist.UnknownFlags = true
 	_ = tmpFlags.Parse(args)
 	if o.storageProvider == "" {
+		if help {
+			cmd.Long = templates.LongDesc(`
+                Note: This help information only shows the common flags for creating a 
+                backup repository, to show provider-specific flags, please specify 
+                the --provider flag. For example:
+
+                    kbcli backuprepo create --provider s3 --help
+            `)
+			return pflag.ErrHelp
+		}
 		return fmt.Errorf("please specify the --%s flag", providerFlagName)
 	}
 
@@ -255,6 +276,17 @@ func (o *createOptions) complete(cmd *cobra.Command) error {
 	return nil
 }
 
+func (o *createOptions) supportedAccessMethods() []string {
+	var methods []string
+	if o.providerObject.Spec.StorageClassTemplate != "" || o.providerObject.Spec.PersistentVolumeClaimTemplate != "" {
+		methods = append(methods, string(dpv1alpha1.AccessMethodMount))
+	}
+	if o.providerObject.Spec.DatasafedConfigTemplate != "" {
+		methods = append(methods, string(dpv1alpha1.AccessMethodTool))
+	}
+	return methods
+}
+
 func (o *createOptions) validate(cmd *cobra.Command) error {
 	// Validate values by the json schema
 	schema := o.providerObject.Spec.ParametersSchema
@@ -275,6 +307,24 @@ func (o *createOptions) validate(cmd *cobra.Command) error {
 		}
 	}
 
+	// Validate access method
+	supportedAccessMethods := o.supportedAccessMethods()
+	if len(supportedAccessMethods) == 0 {
+		return fmt.Errorf("invalid provider \"%s\", it doesn't support any access method", o.storageProvider)
+	}
+	if o.accessMethod != "" && !slices.Contains(supportedAccessMethods, o.accessMethod) {
+		return fmt.Errorf("provider \"%s\" doesn't support \"%s\" access method, supported methods: %q",
+			o.storageProvider, o.accessMethod, supportedAccessMethods)
+	}
+	if o.accessMethod == "" {
+		// Prefer using AccessMethodTool if it's supported
+		if slices.Contains(supportedAccessMethods, string(dpv1alpha1.AccessMethodTool)) {
+			o.accessMethod = string(dpv1alpha1.AccessMethodTool)
+		} else {
+			o.accessMethod = supportedAccessMethods[0]
+		}
+	}
+
 	// Validate pv reclaim policy
 	if !slices.Contains(allowedPVReclaimPolicies, o.pvReclaimPolicy) {
 		return fmt.Errorf("invalid --pv-reclaim-policy \"%s\", the value must be one of %q",
@@ -352,6 +402,7 @@ func (o *createOptions) buildBackupRepoObject(secret *corev1.Secret) (*unstructu
 			Kind:       "BackupRepo",
 		},
 		Spec: dpv1alpha1.BackupRepoSpec{
+			AccessMethod:       dpv1alpha1.AccessMethod(o.accessMethod),
 			StorageProviderRef: o.storageProvider,
 			PVReclaimPolicy:    corev1.PersistentVolumeReclaimPolicy(o.pvReclaimPolicy),
 			VolumeCapacity:     resource.MustParse(o.volumeCapacity),
@@ -441,7 +492,7 @@ func (o *createOptions) run() error {
 
 	// set ownership of the secret to the repo object
 	if createdSecret != nil {
-		_ = o.setSecretOwnership(createdSecret, backupRepoObj)
+		_ = o.setSecretOwnership(createdSecret, createdBackupRepo)
 	}
 
 	printer.PrintLine(fmt.Sprintf("Successfully create backup repo \"%s\".", createdBackupRepo.GetName()))
@@ -454,6 +505,9 @@ func registerFlagCompletionFunc(cmd *cobra.Command, f cmdutil.Factory) {
 		func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 			return utilcomp.CompGetResource(f, util.GVRToString(types.StorageProviderGVR()), toComplete), cobra.ShellCompDirectiveNoFileComp
 		}))
+	util.CheckErr(cmd.RegisterFlagCompletionFunc(
+		"access-method",
+		cobra.FixedCompletions(allowedAccessMethods, cobra.ShellCompDirectiveNoFileComp)))
 	util.CheckErr(cmd.RegisterFlagCompletionFunc(
 		"pv-reclaim-policy",
 		cobra.FixedCompletions(allowedPVReclaimPolicies, cobra.ShellCompDirectiveNoFileComp)))
diff --git a/internal/cli/cmd/backuprepo/create_test.go b/internal/cli/cmd/backuprepo/create_test.go
index 045d0204f94..b47b9a5b1ab 100644
--- a/internal/cli/cmd/backuprepo/create_test.go
+++ b/internal/cli/cmd/backuprepo/create_test.go
@@ -85,7 +85,7 @@ var _ = Describe("backuprepo create command", func() {
 		err := options.init(tf)
 		Expect(err).ShouldNot(HaveOccurred())
 
-		providerObj := testing.FakeStorageProvider("fake-s3")
+		providerObj := testing.FakeStorageProvider("fake-s3", nil)
 		repoObj := testing.FakeBackupRepo("test-backuprepo", false)
 		tf.FakeDynamicClient = fake.NewSimpleDynamicClient(
 			scheme.Scheme, providerObj, repoObj)
@@ -208,7 +208,7 @@ var _ = Describe("backuprepo create command", func() {
 		})
 		It("should validate if there is a default backup repo", func() {
 			By("setting up a default backup repo")
-			providerObj := testing.FakeStorageProvider("fake-s3")
+			providerObj := testing.FakeStorageProvider("fake-s3", nil)
 			repoObj := testing.FakeBackupRepo("test-backuprepo", true)
 			tf.FakeDynamicClient = fake.NewSimpleDynamicClient(
 				scheme.Scheme, providerObj, repoObj)
@@ -220,6 +220,46 @@ var _ = Describe("backuprepo create command", func() {
 			err = options.validate(cmd)
 			Expect(err).Should(MatchError(ContainSubstring("there is already a default backup repo")))
 		})
+		Context("validate access method", func() {
+			const supported = "supported"
+			BeforeEach(func() {
+				options.providerObject.Spec.StorageClassTemplate = ""
+				options.providerObject.Spec.PersistentVolumeClaimTemplate = ""
+				options.providerObject.Spec.DatasafedConfigTemplate = ""
+				options.accessMethod = "" // unspecified
+			})
+			It("should return error if the provider doesn't support any access method", func() {
+				Expect(options.supportedAccessMethods()).Should(BeEmpty())
+				err := options.validate(cmd)
+				Expect(err).Should(MatchError(ContainSubstring("it doesn't support any access method")))
+			})
+			It("should use the mount method if it's the only supported access method", func() {
+				options.providerObject.Spec.StorageClassTemplate = supported
+				err := options.validate(cmd)
+				Expect(err).ShouldNot(HaveOccurred())
+				Expect(options.accessMethod).Should(Equal("Mount"))
+			})
+			It("should use the tool method if it's the only supported access method", func() {
+				options.providerObject.Spec.DatasafedConfigTemplate = supported
+				err := options.validate(cmd)
+				Expect(err).ShouldNot(HaveOccurred())
+				Expect(options.accessMethod).Should(Equal("Tool"))
+			})
+			It("should return error if the specified access method is not supported", func() {
+				options.providerObject.Spec.StorageClassTemplate = supported
+				options.accessMethod = "Tool"
+				err := options.validate(cmd)
+				Expect(err).Should(MatchError(ContainSubstring("doesn't support \"Tool\" access method")))
+			})
+			It("should prefer using the tool method", func() {
+				options.providerObject.Spec.StorageClassTemplate = supported
+				options.providerObject.Spec.DatasafedConfigTemplate = supported
+				options.accessMethod = ""
+				err := options.validate(cmd)
+				Expect(err).ShouldNot(HaveOccurred())
+				Expect(options.accessMethod).Should(Equal("Tool"))
+			})
+		})
 	})
 
 	Describe("run", func() {
@@ -227,7 +267,7 @@ var _ = Describe("backuprepo create command", func() {
 			By("preparing the options")
 			err := options.parseProviderFlags(cmd, []string{
 				"--provider", "fake-s3", "--access-key-id", "abc", "--secret-access-key", "def",
-				"--region", "us-west-1", "--bucket", "test-bucket", "--default",
+				"--region", "us-west-1", "--bucket", "test-bucket", "--default", "--access-method", "Mount",
 			}, tf)
 			Expect(err).ShouldNot(HaveOccurred())
 			err = options.complete(cmd)
diff --git a/internal/cli/cmd/backuprepo/describe.go b/internal/cli/cmd/backuprepo/describe.go
index 918e3451e71..a8d4367a4fa 100644
--- a/internal/cli/cmd/backuprepo/describe.go
+++ b/internal/cli/cmd/backuprepo/describe.go
@@ -68,7 +68,7 @@ func newDescribeCmd(f cmdutil.Factory, streams genericiooptions.IOStreams) *cobr
 	}
 	cmd := &cobra.Command{
 		Use:               "describe",
-		Short:             "Describe a backuprepo.",
+		Short:             "Describe a backup repository.",
 		Example:           describeExample,
 		ValidArgsFunction: util.ResourceNameCompletionFunc(f, types.BackupRepoGVR()),
 		Run: func(cmd *cobra.Command, args []string) {
@@ -127,23 +127,22 @@ func (o *describeBackupRepoOptions) printBackupRepo(backupRepo *dpv1alpha1.Backu
 	printer.PrintLine("Summary:")
 	printer.PrintPairStringToLine("Name", backupRepo.Name)
 	printer.PrintPairStringToLine("Provider", backupRepo.Spec.StorageProviderRef)
-	printer.PrintPairStringToLine("Bucket", backupRepo.Spec.Config["bucket"])
-	if backupRepo.Spec.StorageProviderRef == "minio" {
-		printer.PrintPairStringToLine("Endpoint", backupRepo.Spec.Config["endpoint"])
-	} else {
-		printer.PrintPairStringToLine("Region", backupRepo.Spec.Config["region"])
-	}
 	backups, backupSize, err := countBackupNumsAndSize(o.dynamic, backupRepo)
 	if err != nil {
 		return err
 	}
 	printer.PrintPairStringToLine("Backups", fmt.Sprintf("%d", backups))
-	printer.PrintPairStringToLine("Total data size", backupSize)
+	printer.PrintPairStringToLine("Total Data Size", backupSize)
 
 	printer.PrintLine("\nSpec:")
+	printer.PrintPairStringToLine("AccessMethod", string(backupRepo.Spec.AccessMethod))
 	printer.PrintPairStringToLine("PvReclaimPolicy", string(backupRepo.Spec.PVReclaimPolicy))
 	printer.PrintPairStringToLine("StorageProviderRef", backupRepo.Spec.StorageProviderRef)
 	printer.PrintPairStringToLine("VolumeCapacity", backupRepo.Spec.VolumeCapacity.String())
+	printer.PrintLine("  Config:")
+	for k, v := range backupRepo.Spec.Config {
+		printer.PrintPairStringToLine(k, v, 4)
+	}
 
 	printer.PrintLine("\nStatus:")
 	printer.PrintPairStringToLine("Phase", string(backupRepo.Status.Phase))
diff --git a/internal/cli/cmd/backuprepo/list.go b/internal/cli/cmd/backuprepo/list.go
index 4842ed59cd9..e9b3823b628 100644
--- a/internal/cli/cmd/backuprepo/list.go
+++ b/internal/cli/cmd/backuprepo/list.go
@@ -40,7 +40,7 @@ import (
 
 var (
 	listExample = templates.Examples(`
-	# List all backuprepos
+	# List all backup repositories
 	kbcli backuprepo list
 	`)
 )
@@ -56,7 +56,7 @@ func newListCommand(f cmdutil.Factory, streams genericiooptions.IOStreams) *cobr
 	}
 	cmd := &cobra.Command{
 		Use:               "list",
-		Short:             "List BackupRepo.",
+		Short:             "List Backup Repositories.",
 		Aliases:           []string{"ls"},
 		Example:           listExample,
 		ValidArgsFunction: util.ResourceNameCompletionFunc(f, types.BackupRepoGVR()),
@@ -99,7 +99,7 @@ func printBackupRepoList(o *listBackupRepoOptions) error {
 	}
 
 	if len(infos) == 0 {
-		fmt.Fprintln(o.IOStreams.Out, "No backup Repo found")
+		fmt.Fprintln(o.IOStreams.Out, "No backup repository found")
 		return nil
 	}
 
@@ -134,6 +134,7 @@ func printBackupRepoList(o *listBackupRepoOptions) error {
 			tbl.AddRow(backupRepo.Name,
 				backupRepo.Status.Phase,
 				backupRepo.Spec.StorageProviderRef,
+				backupRepo.Spec.AccessMethod,
 				backupRepo.Status.IsDefault,
 				fmt.Sprintf("%d", backups),
 				backupSize,
@@ -143,7 +144,7 @@ func printBackupRepoList(o *listBackupRepoOptions) error {
 	}
 
 	if err = printer.PrintTable(o.Out, nil, printRows,
-		"NAME", "STATUS", "STORAGE-PROVIDER", "DEFAULT", "BACKUPS", "TOTAL-SIZE"); err != nil {
+		"NAME", "STATUS", "STORAGE-PROVIDER", "ACCESS-METHOD", "DEFAULT", "BACKUPS", "TOTAL-SIZE"); err != nil {
 		return err
 	}
 	return nil
diff --git a/internal/cli/cmd/cluster/dataprotection.go b/internal/cli/cmd/cluster/dataprotection.go
index 14c1510a1a5..206bed076fc 100644
--- a/internal/cli/cmd/cluster/dataprotection.go
+++ b/internal/cli/cmd/cluster/dataprotection.go
@@ -998,7 +998,12 @@ func (o *DescribeBackupOptions) printBackupObj(obj *dpv1alpha1.Backup) error {
 	if obj.Status.BackupMethod != nil {
 		realPrintPairStringToLine("ActionSet Name", obj.Status.BackupMethod.ActionSetName)
 	}
-	realPrintPairStringToLine("PVC Name", obj.Status.PersistentVolumeClaimName)
+	if obj.Status.BackupRepoName != "" {
+		realPrintPairStringToLine("Repository", obj.Status.BackupRepoName)
+	}
+	if obj.Status.PersistentVolumeClaimName != "" {
+		realPrintPairStringToLine("PVC Name", obj.Status.PersistentVolumeClaimName)
+	}
 	if obj.Status.Duration != nil {
 		realPrintPairStringToLine("Duration", duration.HumanDuration(obj.Status.Duration.Duration))
 	}
diff --git a/internal/cli/testing/fake.go b/internal/cli/testing/fake.go
index fe06dd34e1a..c02cb798701 100644
--- a/internal/cli/testing/fake.go
+++ b/internal/cli/testing/fake.go
@@ -992,7 +992,7 @@ func FakeEventForObject(name string, namespace string, object string) *corev1.Ev
 	}
 }
 
-func FakeStorageProvider(name string) *storagev1alpha1.StorageProvider {
+func FakeStorageProvider(name string, mutateFunc func(obj *storagev1alpha1.StorageProvider)) *storagev1alpha1.StorageProvider {
 	storageProvider := &storagev1alpha1.StorageProvider{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: fmt.Sprintf("%s/%s", types.StorageAPIGroup, types.StorageAPIVersion),
@@ -1040,6 +1040,9 @@ mountOptions: {{ index .Parameters "mountOptions" | default "" }}
 		},
 	}
 	storageProvider.SetCreationTimestamp(metav1.Now())
+	if mutateFunc != nil {
+		mutateFunc(storageProvider)
+	}
 	return storageProvider
 }
 
diff --git a/internal/dataprotection/backup/deleter.go b/internal/dataprotection/backup/deleter.go
index c2eb3f5d585..1e863e45b7a 100644
--- a/internal/dataprotection/backup/deleter.go
+++ b/internal/dataprotection/backup/deleter.go
@@ -90,21 +90,35 @@ func (d *Deleter) DeleteBackupFiles(backup *dpv1alpha1.Backup) (DeletionStatus,
 		return DeletionStatusDeleting, nil
 	}
 
-	// if deletion job not exists, create it
-	pvcName := backup.Status.PersistentVolumeClaimName
-	if pvcName == "" {
-		d.Log.Info("skip deleting backup files because PersistentVolumeClaimName is empty",
-			"backup", backup.Name)
-		return DeletionStatusSucceeded, nil
+	var backupRepo *dpv1alpha1.BackupRepo
+	if backup.Status.BackupRepoName != "" {
+		backupRepo = &dpv1alpha1.BackupRepo{}
+		if err = d.Client.Get(d.Ctx, client.ObjectKey{Name: backup.Status.BackupRepoName}, backupRepo); err != nil {
+			if apierrors.IsNotFound(err) {
+				return DeletionStatusSucceeded, nil
+			}
+			return DeletionStatusUnknown, err
+		}
 	}
 
-	// check if backup repo PVC exists, if not, skip to delete backup files
-	pvcKey := client.ObjectKey{Namespace: backup.Namespace, Name: pvcName}
-	if err = d.Client.Get(d.Ctx, pvcKey, &corev1.PersistentVolumeClaim{}); err != nil {
-		if apierrors.IsNotFound(err) {
+	// if backupRepo is nil (likely because it's a legacy backup object), check the backup PVC
+	var legacyPVCName string
+	if backupRepo == nil {
+		legacyPVCName = backup.Status.PersistentVolumeClaimName
+		if legacyPVCName == "" {
+			d.Log.Info("skip deleting backup files because PersistentVolumeClaimName is empty",
+				"backup", backup.Name)
 			return DeletionStatusSucceeded, nil
 		}
-		return DeletionStatusUnknown, err
+
+		// check if the backup PVC exists, if not, skip to delete backup files
+		pvcKey := client.ObjectKey{Namespace: backup.Namespace, Name: legacyPVCName}
+		if err = d.Client.Get(d.Ctx, pvcKey, &corev1.PersistentVolumeClaim{}); err != nil {
+			if apierrors.IsNotFound(err) {
+				return DeletionStatusSucceeded, nil
+			}
+			return DeletionStatusUnknown, err
+		}
 	}
 
 	backupFilePath := backup.Status.Path
@@ -117,13 +131,14 @@ func (d *Deleter) DeleteBackupFiles(backup *dpv1alpha1.Backup) (DeletionStatus,
 			"backupFilePath", backupFilePath, "backup", backup.Name)
 		return DeletionStatusSucceeded, nil
 	}
-	return DeletionStatusDeleting, d.createDeleteBackupFilesJob(jobKey, backup, pvcName, backup.Status.Path)
+	return DeletionStatusDeleting, d.createDeleteBackupFilesJob(jobKey, backup, backupRepo, legacyPVCName, backup.Status.Path)
 }
 
 func (d *Deleter) createDeleteBackupFilesJob(
 	jobKey types.NamespacedName,
 	backup *dpv1alpha1.Backup,
-	backupPVCName string,
+	backupRepo *dpv1alpha1.BackupRepo,
+	legacyPVCName string,
 	backupFilePath string) error {
 	// make sure the path has a leading slash
 	if !strings.HasPrefix(backupFilePath, "/") {
@@ -133,30 +148,31 @@ func (d *Deleter) createDeleteBackupFilesJob(
 	// this script first deletes the directory where the backup is located (including files
 	// in the directory), and then traverses up the path level by level to clean up empty directories.
 	deleteScript := fmt.Sprintf(`
-		backupPathBase=%s;
-		targetPath="${backupPathBase}%s";
+set -x
+export PATH="$PATH:$%s";
+targetPath="%s";
 
-		echo "removing backup files in ${targetPath}";
-		rm -rf "${targetPath}";
+echo "removing backup files in ${targetPath}";
+datasafed rm -r "${targetPath}";
 
-		absBackupPathBase=$(realpath "${backupPathBase}");
-		curr=$(realpath "${targetPath}");
-		while true; do
-			parent=$(dirname "${curr}");
-			if [ "${parent}" == "${absBackupPathBase}" ]; then
-				echo "reach backupPathBase ${backupPathBase}, done";
-				break;
-			fi;
-			if [ ! "$(ls -A "${parent}")" ]; then
-				echo "${parent} is empty, removing it...";
-				rmdir "${parent}";
-			else
-				echo "${parent} is not empty, done";
-				break;
-			fi;
-			curr="${parent}";
-		done
-	`, RepoVolumeMountPath, backupFilePath)
+curr="${targetPath}";
+while true; do
+	parent=$(dirname "${curr}");
+	if [ "${parent}" == "/" ]; then
+		echo "reach to root, done";
+		break;
+	fi;
+	result=$(datasafed list "${parent}");
+	if [ -z "$result" ]; then
+		echo "${parent} is empty, removing it...";
+		datasafed rmdir "${parent}";
+	else
+		echo "${parent} is not empty, done";
+		break;
+	fi;
+	curr="${parent}";
+done
+	`, dptypes.DPDatasafedBinPath, backupFilePath)
 
 	runAsUser := int64(0)
 	container := corev1.Container{
@@ -169,9 +185,6 @@ func (d *Deleter) createDeleteBackupFilesJob(
 			AllowPrivilegeEscalation: boolptr.False(),
 			RunAsUser:                &runAsUser,
 		},
-		VolumeMounts: []corev1.VolumeMount{
-			buildBackupRepoVolumeMount(backupPVCName),
-		},
 	}
 	ctrlutil.InjectZeroResourcesLimitsIfEmpty(&container)
 
@@ -179,14 +192,15 @@ func (d *Deleter) createDeleteBackupFilesJob(
 	podSpec := corev1.PodSpec{
 		Containers:    []corev1.Container{container},
 		RestartPolicy: corev1.RestartPolicyNever,
-		Volumes: []corev1.Volume{
-			buildBackupRepoVolume(backupPVCName),
-		},
 	}
-
 	if err := utils.AddTolerations(&podSpec); err != nil {
 		return err
 	}
+	if backupRepo != nil {
+		utils.InjectDatasafed(&podSpec, backupRepo, RepoVolumeMountPath, backupFilePath)
+	} else {
+		utils.InjectDatasafedWithPVC(&podSpec, legacyPVCName, RepoVolumeMountPath, backupFilePath)
+	}
 
 	// build job
 	job := &batchv1.Job{
diff --git a/internal/dataprotection/backup/request.go b/internal/dataprotection/backup/request.go
index c19bdcb91e0..5a967dc3235 100644
--- a/internal/dataprotection/backup/request.go
+++ b/internal/dataprotection/backup/request.go
@@ -38,11 +38,13 @@ import (
 )
 
 const (
-	BackupDataJobNamePrefix   = "dp-backup"
-	prebackupJobNamePrefix    = "dp-prebackup"
-	postbackupJobNamePrefix   = "dp-postbackup"
-	backupDataContainerName   = "backupdata"
-	syncProgressContainerName = "sync-progress"
+	BackupDataJobNamePrefix      = "dp-backup"
+	prebackupJobNamePrefix       = "dp-prebackup"
+	postbackupJobNamePrefix      = "dp-postbackup"
+	backupDataContainerName      = "backupdata"
+	syncProgressContainerName    = "sync-progress"
+	syncProgressSharedVolumeName = "sync-progress-shared-volume"
+	syncProgressSharedMountPath  = "/dp-sync-progress"
 )
 
 // Request is a request for a backup, with all references to other objects.
@@ -50,13 +52,14 @@ type Request struct {
 	*dpv1alpha1.Backup
 	intctrlutil.RequestCtx
 
-	Client        client.Client
-	BackupPolicy  *dpv1alpha1.BackupPolicy
-	BackupMethod  *dpv1alpha1.BackupMethod
-	ActionSet     *dpv1alpha1.ActionSet
-	TargetPods    []*corev1.Pod
-	BackupRepoPVC *corev1.PersistentVolumeClaim
-	BackupRepo    *dpv1alpha1.BackupRepo
+	Client           client.Client
+	BackupPolicy     *dpv1alpha1.BackupPolicy
+	BackupMethod     *dpv1alpha1.BackupMethod
+	ActionSet        *dpv1alpha1.ActionSet
+	TargetPods       []*corev1.Pod
+	BackupRepoPVC    *corev1.PersistentVolumeClaim
+	BackupRepo       *dpv1alpha1.BackupRepo
+	ToolConfigSecret *corev1.Secret
 }
 
 func (r *Request) GetBackupType() string {
@@ -273,14 +276,18 @@ func (r *Request) buildJobActionPodSpec(name string, job *dpv1alpha1.JobActionSp
 				Name:  dptypes.DPBackupName,
 				Value: r.Backup.Name,
 			},
-			{
-				Name:  dptypes.DPBackupDIR,
-				Value: buildBackupPathInContainer(r.Backup, r.BackupPolicy.Spec.PathPrefix),
-			},
 			{
 				Name:  dptypes.DPTargetPodName,
 				Value: targetPod.Name,
 			},
+			{
+				Name:  dptypes.DPBackupBasePath,
+				Value: BuildBackupPath(r.Backup, r.BackupPolicy.Spec.PathPrefix),
+			},
+			{
+				Name:  dptypes.DPBackupInfoFile,
+				Value: syncProgressSharedMountPath + "/" + backupInfoFileName,
+			},
 			{
 				Name:  dptypes.DPTTL,
 				Value: r.Spec.RetentionPeriod.String(),
@@ -294,15 +301,27 @@ func (r *Request) buildJobActionPodSpec(name string, job *dpv1alpha1.JobActionSp
 	}
 
 	buildVolumes := func() []corev1.Volume {
-		return append([]corev1.Volume{
-			buildBackupRepoVolume(r.BackupRepoPVC.Name),
-		}, getVolumesByVolumeInfo(targetPod, r.BackupMethod.TargetVolumes)...)
+		return append(
+			[]corev1.Volume{
+				{
+					Name: syncProgressSharedVolumeName,
+					VolumeSource: corev1.VolumeSource{
+						EmptyDir: &corev1.EmptyDirVolumeSource{},
+					},
+				},
+			},
+			getVolumesByVolumeInfo(targetPod, r.BackupMethod.TargetVolumes)...)
 	}
 
 	buildVolumeMounts := func() []corev1.VolumeMount {
-		return append([]corev1.VolumeMount{
-			buildBackupRepoVolumeMount(r.BackupRepoPVC.Name),
-		}, getVolumeMountsByVolumeInfo(targetPod, r.BackupMethod.TargetVolumes)...)
+		return append(
+			[]corev1.VolumeMount{
+				{
+					Name:      syncProgressSharedVolumeName,
+					MountPath: syncProgressSharedMountPath,
+				},
+			},
+			getVolumeMountsByVolumeInfo(targetPod, r.BackupMethod.TargetVolumes)...)
 	}
 
 	runAsUser := int64(0)
@@ -348,6 +367,8 @@ func (r *Request) buildJobActionPodSpec(name string, job *dpv1alpha1.JobActionSp
 			corev1.LabelHostname: targetPod.Spec.NodeName,
 		}
 	}
+	utils.InjectDatasafed(podSpec, r.BackupRepo, RepoVolumeMountPath,
+		BuildBackupPath(r.Backup, r.BackupPolicy.Spec.PathPrefix))
 	return podSpec
 }
 
@@ -373,10 +394,6 @@ func (r *Request) injectSyncProgressContainer(podSpec *corev1.PodSpec,
 		checkIntervalSeconds = *sync.IntervalSeconds
 	}
 	container.Env = append(container.Env,
-		corev1.EnvVar{
-			Name:  dptypes.DPBackupInfoFile,
-			Value: buildBackupInfoFilePath(r.Backup, r.BackupPolicy.Spec.PathPrefix),
-		},
 		corev1.EnvVar{
 			Name:  dptypes.DPCheckInterval,
 			Value: fmt.Sprintf("%d", checkIntervalSeconds)},
diff --git a/internal/dataprotection/backup/utils.go b/internal/dataprotection/backup/utils.go
index 95cf776bd90..89bd6292554 100644
--- a/internal/dataprotection/backup/utils.go
+++ b/internal/dataprotection/backup/utils.go
@@ -134,28 +134,6 @@ func getPVCsByVolumeNames(cli client.Client,
 	return all, nil
 }
 
-func GenerateBackupRepoVolumeName(pvcName string) string {
-	return fmt.Sprintf("dp-backup-%s", pvcName)
-}
-
-func buildBackupRepoVolume(pvcName string) corev1.Volume {
-	return corev1.Volume{
-		Name: GenerateBackupRepoVolumeName(pvcName),
-		VolumeSource: corev1.VolumeSource{
-			PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
-				ClaimName: pvcName,
-			},
-		},
-	}
-}
-
-func buildBackupRepoVolumeMount(pvcName string) corev1.VolumeMount {
-	return corev1.VolumeMount{
-		Name:      GenerateBackupRepoVolumeName(pvcName),
-		MountPath: RepoVolumeMountPath,
-	}
-}
-
 func excludeLabelsForWorkload() []string {
 	return []string{constant.KBAppComponentLabelKey}
 }
@@ -208,14 +186,6 @@ func generateUniqueNameWithBackupSchedule(backupSchedule *dpv1alpha1.BackupSched
 	return uniqueName
 }
 
-func buildBackupInfoFilePath(backup *dpv1alpha1.Backup, pathPrefix string) string {
-	return buildBackupPathInContainer(backup, pathPrefix) + "/" + backupInfoFileName
-}
-
-func buildBackupPathInContainer(backup *dpv1alpha1.Backup, pathPrefix string) string {
-	return RepoVolumeMountPath + BuildBackupPath(backup, pathPrefix)
-}
-
 // BuildBackupPath builds the path to storage backup datas in backup repository.
 func BuildBackupPath(backup *dpv1alpha1.Backup, pathPrefix string) string {
 	pathPrefix = strings.TrimRight(pathPrefix, "/")
diff --git a/internal/dataprotection/errors/errors.go b/internal/dataprotection/errors/errors.go
index 207f374bd32..8d2590a7617 100644
--- a/internal/dataprotection/errors/errors.go
+++ b/internal/dataprotection/errors/errors.go
@@ -33,6 +33,10 @@ const (
 	ErrorTypeBackupNotCompleted intctrlutil.ErrorType = "BackupNotCompleted"
 	// ErrorTypeBackupPVCNameIsEmpty pvc name for backup is empty
 	ErrorTypeBackupPVCNameIsEmpty intctrlutil.ErrorType = "BackupPVCNameIsEmpty"
+	// ErrorTypeBackupRepoIsNotReady the backup repository is not ready
+	ErrorTypeBackupRepoIsNotReady intctrlutil.ErrorType = "BackupRepoIsNotReady"
+	// ErrorTypeToolConfigSecretNameIsEmpty the name of  repository is not ready
+	ErrorTypeToolConfigSecretNameIsEmpty intctrlutil.ErrorType = "ToolConfigSecretNameIsEmpty"
 	// ErrorTypeBackupJobFailed backup job failed
 	ErrorTypeBackupJobFailed intctrlutil.ErrorType = "BackupJobFailed"
 	// ErrorTypeStorageNotMatch storage not match
@@ -59,6 +63,16 @@ func NewBackupPVTemplateNotFound(cmName, cmNamespace string) *intctrlutil.Error
 	return intctrlutil.NewErrorf(ErrorTypeBackupPVTemplateNotFound, `"the persistentVolume template is empty in the configMap %s/%s", pvConfig.Namespace, pvConfig.Name`, cmNamespace, cmName)
 }
 
+// NewBackupRepoIsNotReady returns a new Error with ErrorTypeBackupRepoIsNotReady.
+func NewBackupRepoIsNotReady(backupRepo string) *intctrlutil.Error {
+	return intctrlutil.NewErrorf(ErrorTypeBackupRepoIsNotReady, `the backup repository %s is not ready`, backupRepo)
+}
+
+// NewToolConfigSecretNameIsEmpty returns a new Error with ErrorTypeToolConfigSecretNameIsEmpty.
+func NewToolConfigSecretNameIsEmpty(backupRepo string) *intctrlutil.Error {
+	return intctrlutil.NewErrorf(ErrorTypeToolConfigSecretNameIsEmpty, `the secret name of tool config from %s is empty`, backupRepo)
+}
+
 // NewBackupPVCNameIsEmpty returns a new Error with ErrorTypeBackupPVCNameIsEmpty.
 func NewBackupPVCNameIsEmpty(backupRepo, backupPolicyName string) *intctrlutil.Error {
 	return intctrlutil.NewErrorf(ErrorTypeBackupPVCNameIsEmpty, `the persistentVolumeClaim name of %s is empty in BackupPolicy "%s"`, backupRepo, backupPolicyName)
diff --git a/internal/dataprotection/errors/errors_test.go b/internal/dataprotection/errors/errors_test.go
index fde1dcf9335..ccfd3e42ccd 100644
--- a/internal/dataprotection/errors/errors_test.go
+++ b/internal/dataprotection/errors/errors_test.go
@@ -61,10 +61,18 @@ func TestNewErrors(t *testing.T) {
 	if !intctrlutil.IsTargetError(pvTemplateNotFound, ErrorTypeBackupPVTemplateNotFound) {
 		t.Error("should be error of BackupPVTemplateNotFound")
 	}
-	pvsIsEmpty := NewBackupPVCNameIsEmpty("datafile", "policy-test1")
-	if !intctrlutil.IsTargetError(pvsIsEmpty, ErrorTypeBackupPVCNameIsEmpty) {
+	pvcIsEmpty := NewBackupPVCNameIsEmpty("datafile", "policy-test1")
+	if !intctrlutil.IsTargetError(pvcIsEmpty, ErrorTypeBackupPVCNameIsEmpty) {
 		t.Error("should be error of BackupPVCNameIsEmpty")
 	}
+	repoIsNotReady := NewBackupRepoIsNotReady("repo")
+	if !intctrlutil.IsTargetError(repoIsNotReady, ErrorTypeBackupRepoIsNotReady) {
+		t.Error("should be error of BackupRepoIsNotReady")
+	}
+	toolConfigSecretNameIsEmpty := NewToolConfigSecretNameIsEmpty("repo")
+	if !intctrlutil.IsTargetError(toolConfigSecretNameIsEmpty, ErrorTypeToolConfigSecretNameIsEmpty) {
+		t.Error("should be error of ToolConfigSecretNameIsEmpty")
+	}
 	jobFailed := NewBackupJobFailed("jobName")
 	if !intctrlutil.IsTargetError(jobFailed, ErrorTypeBackupJobFailed) {
 		t.Error("should be error of BackupJobFailed")
diff --git a/internal/dataprotection/restore/builder.go b/internal/dataprotection/restore/builder.go
index 3ab76ef42ac..b2a56da72a6 100644
--- a/internal/dataprotection/restore/builder.go
+++ b/internal/dataprotection/restore/builder.go
@@ -40,6 +40,8 @@ type restoreJobBuilder struct {
 	restore            *dpv1alpha1.Restore
 	stage              dpv1alpha1.RestoreStage
 	backupSet          BackupActionSet
+	backupRepo         *dpv1alpha1.BackupRepo
+	buildWithRepo      bool
 	env                []corev1.EnvVar
 	commonVolumes      []corev1.Volume
 	commonVolumeMounts []corev1.VolumeMount
@@ -55,10 +57,11 @@ type restoreJobBuilder struct {
 	labels               map[string]string
 }
 
-func newRestoreJobBuilder(restore *dpv1alpha1.Restore, backupSet BackupActionSet, stage dpv1alpha1.RestoreStage) *restoreJobBuilder {
+func newRestoreJobBuilder(restore *dpv1alpha1.Restore, backupSet BackupActionSet, backupRepo *dpv1alpha1.BackupRepo, stage dpv1alpha1.RestoreStage) *restoreJobBuilder {
 	return &restoreJobBuilder{
 		restore:            restore,
 		backupSet:          backupSet,
+		backupRepo:         backupRepo,
 		stage:              stage,
 		commonVolumes:      []corev1.Volume{},
 		commonVolumeMounts: []corev1.VolumeMount{},
@@ -93,24 +96,6 @@ func (r *restoreJobBuilder) buildPVCVolumeAndMount(
 		claim.VolumeSource, r.backupSet.Backup.Name))
 }
 
-// addBackupVolumeAndMount adds the volume and volumeMount of backup pvc to common volumes and volumeMounts slice.
-func (r *restoreJobBuilder) addBackupVolumeAndMount() *restoreJobBuilder {
-	if r.backupSet.Backup.Status.PersistentVolumeClaimName != "" {
-		backupName := r.backupSet.Backup.Name
-		r.commonVolumes = append(r.commonVolumes, corev1.Volume{
-			Name: backupName,
-			VolumeSource: corev1.VolumeSource{
-				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: r.backupSet.Backup.Status.PersistentVolumeClaimName},
-			},
-		})
-		r.commonVolumeMounts = append(r.commonVolumeMounts, corev1.VolumeMount{
-			Name:      backupName,
-			MountPath: "/" + backupName,
-		})
-	}
-	return r
-}
-
 // addToCommonVolumesAndMounts adds the volume and volumeMount to common volumes and volumeMounts slice.
 func (r *restoreJobBuilder) addToCommonVolumesAndMounts(volume *corev1.Volume, volumeMount *corev1.VolumeMount) *restoreJobBuilder {
 	if volume != nil {
@@ -171,6 +156,11 @@ func (r *restoreJobBuilder) addLabel(key, value string) *restoreJobBuilder {
 	return r
 }
 
+func (r *restoreJobBuilder) attachBackupRepo() *restoreJobBuilder {
+	r.buildWithRepo = true
+	return r
+}
+
 // addCommonEnv adds the common envs for each restore job.
 func (r *restoreJobBuilder) addCommonEnv() *restoreJobBuilder {
 	backupName := r.backupSet.Backup.Name
@@ -179,7 +169,7 @@ func (r *restoreJobBuilder) addCommonEnv() *restoreJobBuilder {
 	// add mount path env of backup dir
 	filePath := r.backupSet.Backup.Status.Path
 	if filePath != "" {
-		r.env = append(r.env, corev1.EnvVar{Name: dptypes.DPBackupDIR, Value: fmt.Sprintf("/%s%s", backupName, filePath)})
+		r.env = append(r.env, corev1.EnvVar{Name: dptypes.DPBackupBasePath, Value: filePath})
 		// TODO: add continuous file path env
 	}
 	// add time env
@@ -305,5 +295,18 @@ func (r *restoreJobBuilder) build() *batchv1.Job {
 	intctrlutil.InjectZeroResourcesLimitsIfEmpty(&container)
 	job.Spec.Template.Spec.Containers = []corev1.Container{container}
 	controllerutil.AddFinalizer(job, dptypes.DataProtectionFinalizerName)
+
+	// 3. inject datasafed if needed
+	if r.buildWithRepo {
+		mountPath := "/backupdata"
+		backupPath := r.backupSet.Backup.Status.Path
+		if r.backupRepo != nil {
+			utils.InjectDatasafed(&job.Spec.Template.Spec, r.backupRepo, mountPath, backupPath)
+		} else if pvcName := r.backupSet.Backup.Status.PersistentVolumeClaimName; pvcName != "" {
+			// If the backup object was created in an old version that doesn't have the backupRepo field,
+			// use the PVC name field as a fallback.
+			utils.InjectDatasafedWithPVC(&job.Spec.Template.Spec, pvcName, mountPath, backupPath)
+		}
+	}
 	return job
 }
diff --git a/internal/dataprotection/restore/manager.go b/internal/dataprotection/restore/manager.go
index d7fce7f00af..cc017795ed5 100644
--- a/internal/dataprotection/restore/manager.go
+++ b/internal/dataprotection/restore/manager.go
@@ -229,6 +229,21 @@ func (r *RestoreManager) RestorePVCFromSnapshot(reqCtx intctrlutil.RequestCtx, c
 	return nil
 }
 
+func (r *RestoreManager) prepareBackupRepo(reqCtx intctrlutil.RequestCtx, cli client.Client, backupSet BackupActionSet) (*dpv1alpha1.BackupRepo, error) {
+	if backupSet.Backup.Status.BackupRepoName != "" {
+		backupRepo := &dpv1alpha1.BackupRepo{}
+		err := cli.Get(reqCtx.Ctx, client.ObjectKey{Name: backupSet.Backup.Status.BackupRepoName}, backupRepo)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				err = intctrlutil.NewFatalError(err.Error())
+			}
+			return nil, err
+		}
+		return backupRepo, nil
+	}
+	return nil, nil
+}
+
 // BuildPrepareDataJobs builds the restore jobs for prepare pvc's data, and will create the target pvcs if not exist.
 func (r *RestoreManager) BuildPrepareDataJobs(reqCtx intctrlutil.RequestCtx, cli client.Client, backupSet BackupActionSet, actionName string) ([]*batchv1.Job, error) {
 	prepareDataConfig := r.Restore.Spec.PrepareDataConfig
@@ -238,11 +253,15 @@ func (r *RestoreManager) BuildPrepareDataJobs(reqCtx intctrlutil.RequestCtx, cli
 	if !backupSet.ActionSet.HasPrepareDataStage() {
 		return nil, nil
 	}
-	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, dpv1alpha1.PrepareData).
+	backupRepo, err := r.prepareBackupRepo(reqCtx, cli, backupSet)
+	if err != nil {
+		return nil, err
+	}
+	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, backupRepo, dpv1alpha1.PrepareData).
 		setImage(backupSet.ActionSet.Spec.Restore.PrepareData.Image).
 		setCommand(backupSet.ActionSet.Spec.Restore.PrepareData.Command).
-		addBackupVolumeAndMount().
-		addCommonEnv()
+		addCommonEnv().
+		attachBackupRepo()
 
 	createPVCIfNotExistsAndBuildVolume := func(claim dpv1alpha1.RestoreVolumeClaim, identifier string) (*corev1.Volume, *corev1.VolumeMount, error) {
 		if err := r.createPVCIfNotExist(reqCtx, cli, claim.ObjectMeta, claim.VolumeClaimSpec); err != nil {
@@ -315,6 +334,8 @@ func (r *RestoreManager) BuildPrepareDataJobs(reqCtx intctrlutil.RequestCtx, cli
 }
 
 func (r *RestoreManager) BuildVolumePopulateJob(
+	reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
 	backupSet BackupActionSet,
 	populatePVC *corev1.PersistentVolumeClaim,
 	index int) (*batchv1.Job, error) {
@@ -325,12 +346,16 @@ func (r *RestoreManager) BuildVolumePopulateJob(
 	if !backupSet.ActionSet.HasPrepareDataStage() {
 		return nil, nil
 	}
-	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, dpv1alpha1.PrepareData).
+	backupRepo, err := r.prepareBackupRepo(reqCtx, cli, backupSet)
+	if err != nil {
+		return nil, err
+	}
+	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, backupRepo, dpv1alpha1.PrepareData).
 		setJobName(fmt.Sprintf("%s-%d", populatePVC.Name, index)).
 		addLabel(DataProtectionLabelPopulatePVCKey, populatePVC.Name).
 		setImage(backupSet.ActionSet.Spec.Restore.PrepareData.Image).
 		setCommand(backupSet.ActionSet.Spec.Restore.PrepareData.Command).
-		addBackupVolumeAndMount().
+		attachBackupRepo().
 		addCommonEnv()
 	volume, volumeMount, err := jobBuilder.buildPVCVolumeAndMount(*prepareDataConfig.DataSourceRef, populatePVC.Name, "dp-claim")
 	if err != nil {
@@ -349,6 +374,10 @@ func (r *RestoreManager) BuildPostReadyActionJobs(reqCtx intctrlutil.RequestCtx,
 	if !backupSet.ActionSet.HasPostReadyStage() {
 		return nil, nil
 	}
+	backupRepo, err := r.prepareBackupRepo(reqCtx, cli, backupSet)
+	if err != nil {
+		return nil, err
+	}
 	actionSpec := backupSet.ActionSet.Spec.Restore.PostReady[step]
 	getTargetPodList := func(labelSelector metav1.LabelSelector, msgKey string) ([]corev1.Pod, error) {
 		targetPodList, err := utils.GetPodListByLabelSelector(reqCtx, cli, labelSelector)
@@ -361,7 +390,7 @@ func (r *RestoreManager) BuildPostReadyActionJobs(reqCtx intctrlutil.RequestCtx,
 		return targetPodList.Items, nil
 	}
 
-	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, dpv1alpha1.PostReady).addCommonEnv()
+	jobBuilder := newRestoreJobBuilder(r.Restore, backupSet, backupRepo, dpv1alpha1.PostReady).addCommonEnv()
 
 	buildJobName := func(index int) string {
 		jobName := fmt.Sprintf("restore-post-ready-%s-%s-%d-%d", r.Restore.UID[:8], backupSet.Backup.Name, step, index)
@@ -391,7 +420,7 @@ func (r *RestoreManager) BuildPostReadyActionJobs(reqCtx intctrlutil.RequestCtx,
 		}
 		job := jobBuilder.setImage(actionSpec.Job.Image).
 			setJobName(buildJobName(0)).
-			addBackupVolumeAndMount().
+			attachBackupRepo().
 			setCommand(actionSpec.Job.Command).
 			setToleration(targetPod.Spec.Tolerations).
 			addTargetPodAndCredentialEnv(&targetPod, r.Restore.Spec.ReadyConfig.ConnectionCredential).
diff --git a/internal/dataprotection/restore/manager_test.go b/internal/dataprotection/restore/manager_test.go
index 2d27c7507b7..04aa5bfb2dc 100644
--- a/internal/dataprotection/restore/manager_test.go
+++ b/internal/dataprotection/restore/manager_test.go
@@ -262,7 +262,7 @@ var _ = Describe("Backup Deleter Test", func() {
 					Name: "test-populate-pvc",
 				},
 			}
-			job, err := restoreMGR.BuildVolumePopulateJob(*backupSet, populatePVC, 0)
+			job, err := restoreMGR.BuildVolumePopulateJob(reqCtx, k8sClient, *backupSet, populatePVC, 0)
 			Expect(err).ShouldNot(HaveOccurred())
 			Expect(job).ShouldNot(BeNil())
 		})
diff --git a/internal/dataprotection/types/constant.go b/internal/dataprotection/types/constant.go
index 96756abc1a3..a0ef0667637 100644
--- a/internal/dataprotection/types/constant.go
+++ b/internal/dataprotection/types/constant.go
@@ -84,8 +84,8 @@ const (
 	DPDBPort = "DP_DB_PORT"
 	// DPTargetPodName the target pod name
 	DPTargetPodName = "DP_TARGET_POD_NAME"
-	// DPBackupDIR the dest directory for backup data
-	DPBackupDIR = "DP_BACKUP_DIR"
+	// DPBackupBasePath the base path for backup data in the storage
+	DPBackupBasePath = "DP_BACKUP_BASE_PATH"
 	// DPBackupName backup CR name
 	DPBackupName = "DP_BACKUP_NAME"
 	// DPTTL backup time to live, reference the backupPolicy.spec.retention.ttl
@@ -108,6 +108,10 @@ const (
 	DPBaseBackupStartTimestamp = "BASE_BACKUP_START_TIMESTAMP" // base backup start timestamp for pitr
 	// DPBackupStopTime backup stop time
 	DPBackupStopTime = "BACKUP_STOP_TIME" // backup stop time
+	// DPDatasafedLocalBackendPath force datasafed to use local backend with the path
+	DPDatasafedLocalBackendPath = "DATASAFED_LOCAL_BACKEND_PATH"
+	// DPDatasafedBinPath the path containing the datasafed binary
+	DPDatasafedBinPath = "DP_DATASAFED_BIN_PATH"
 )
 
 const (
diff --git a/internal/dataprotection/utils/backuprepo.go b/internal/dataprotection/utils/backuprepo.go
new file mode 100644
index 00000000000..939908c0ccc
--- /dev/null
+++ b/internal/dataprotection/utils/backuprepo.go
@@ -0,0 +1,137 @@
+/*
+Copyright (C) 2022-2023 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 utils
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+	"github.com/apecloud/kubeblocks/internal/constant"
+	dptypes "github.com/apecloud/kubeblocks/internal/dataprotection/types"
+	viper "github.com/apecloud/kubeblocks/internal/viperx"
+)
+
+const (
+	datasafedImageEnv        = "DATASAFED_IMAGE"
+	defaultDatasafedImage    = "apecloud/datasafed:latest"
+	datasafedBinMountPath    = "/bin/datasafed"
+	datasafedConfigMountPath = "/etc/datasafed"
+)
+
+func InjectDatasafed(podSpec *corev1.PodSpec, repo *dpv1alpha1.BackupRepo, repoVolumeMountPath string, backupPath string) {
+	if repo.AccessByMount() {
+		InjectDatasafedWithPVC(podSpec, repo.Status.BackupPVCName, repoVolumeMountPath, backupPath)
+	} else if repo.AccessByTool() {
+		InjectDatasafedWithConfig(podSpec, repo.Status.ToolConfigSecretName, backupPath)
+	}
+}
+
+func InjectDatasafedWithPVC(podSpec *corev1.PodSpec, pvcName string, mountPath string, backupPath string) {
+	volumeName := "dp-backup-data"
+	volume := corev1.Volume{
+		Name: volumeName,
+		VolumeSource: corev1.VolumeSource{
+			PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+				ClaimName: pvcName,
+			},
+		},
+	}
+	volumeMount := corev1.VolumeMount{
+		Name:      volumeName,
+		MountPath: mountPath,
+	}
+	envs := []corev1.EnvVar{
+		{
+			// force datasafed to use local backend with the path
+			Name:  dptypes.DPDatasafedLocalBackendPath,
+			Value: mountPath,
+		},
+	}
+	injectElements(podSpec, toSlice(volume), toSlice(volumeMount), envs)
+	injectDatasafedInstaller(podSpec)
+}
+
+func InjectDatasafedWithConfig(podSpec *corev1.PodSpec, configSecretName string, backupPath string) {
+	volumeName := "dp-datasafed-config"
+	volume := corev1.Volume{
+		Name: volumeName,
+		VolumeSource: corev1.VolumeSource{
+			Secret: &corev1.SecretVolumeSource{
+				SecretName: configSecretName,
+			},
+		},
+	}
+	volumeMount := corev1.VolumeMount{
+		Name:      volumeName,
+		ReadOnly:  true,
+		MountPath: datasafedConfigMountPath,
+	}
+	injectElements(podSpec, toSlice(volume), toSlice(volumeMount), nil)
+	injectDatasafedInstaller(podSpec)
+}
+
+func injectDatasafedInstaller(podSpec *corev1.PodSpec) {
+	sharedVolumeName := "dp-datasafed-bin"
+	sharedVolume := corev1.Volume{
+		Name: sharedVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			EmptyDir: &corev1.EmptyDirVolumeSource{},
+		},
+	}
+	sharedVolumeMount := corev1.VolumeMount{
+		Name:      sharedVolumeName,
+		MountPath: datasafedBinMountPath,
+	}
+	env := corev1.EnvVar{
+		Name:  dptypes.DPDatasafedBinPath,
+		Value: datasafedBinMountPath,
+	}
+
+	// copy the datasafed binary from the image to the shared volume
+	datasafedImage := viper.GetString(datasafedImageEnv)
+	if datasafedImage == "" {
+		datasafedImage = defaultDatasafedImage
+	}
+	initContainer := corev1.Container{
+		Name:            "dp-copy-datasafed",
+		Image:           datasafedImage,
+		ImagePullPolicy: corev1.PullPolicy(viper.GetString(constant.KBImagePullPolicy)),
+		Command:         []string{"/bin/sh", "-c", fmt.Sprintf("/scripts/install-datasafed.sh %s", datasafedBinMountPath)},
+		VolumeMounts:    []corev1.VolumeMount{sharedVolumeMount},
+	}
+
+	podSpec.InitContainers = append(podSpec.InitContainers, initContainer)
+	injectElements(podSpec, toSlice(sharedVolume), toSlice(sharedVolumeMount), toSlice(env))
+}
+
+func injectElements(podSpec *corev1.PodSpec, volumes []corev1.Volume, volumeMounts []corev1.VolumeMount, envs []corev1.EnvVar) {
+	podSpec.Volumes = append(podSpec.Volumes, volumes...)
+	for i := range podSpec.Containers {
+		container := &podSpec.Containers[i]
+		container.VolumeMounts = append(container.VolumeMounts, volumeMounts...)
+		container.Env = append(container.Env, envs...)
+	}
+}
+
+func toSlice[T any](s ...T) []T {
+	return s
+}
diff --git a/internal/testutil/dataprotection/backup_utils.go b/internal/testutil/dataprotection/backup_utils.go
index 320d3f9ef93..82fedeaf15d 100644
--- a/internal/testutil/dataprotection/backup_utils.go
+++ b/internal/testutil/dataprotection/backup_utils.go
@@ -24,6 +24,8 @@ import (
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
@@ -79,6 +81,11 @@ func NewFakeStorageProvider(testCtx *testutil.TestContext,
 	// the storage provider controller is not running, so set the status manually
 	Expect(testapps.ChangeObjStatus(testCtx, sp, func() {
 		sp.Status.Phase = storagev1alpha1.StorageProviderReady
+		meta.SetStatusCondition(&sp.Status.Conditions, metav1.Condition{
+			Type:   storagev1alpha1.ConditionTypeCSIDriverInstalled,
+			Status: metav1.ConditionTrue,
+			Reason: "CSIDriverInstalled",
+		})
 	})).Should(Succeed())
 	return sp
 }