Merge pull request #10140 from Lyndon-Li/set-cbt-service-to-uploader

Set CBT service to uploader

pkg/cbtservice/csi_service_impl.go

@@ -86,6 +86,12 @@ func (s *ServiceImpl) GetAllocatedBlocks(ctx context.Context, snapshot string, r
 		return err
 	}
 
+	saNamespace := ""
+	if s.SAName != "" {
+		// The SA is created in the same namespace as Velero server. vsNamespace is the namespace of Velero	server.
+		saNamespace = s.vsNamespace
+	}
+
 	args := iterator.Args{
 		SnapshotName: snapshot,
 		Emitter: &emitterImpl{
@@ -95,7 +101,7 @@ func (s *ServiceImpl) GetAllocatedBlocks(ctx context.Context, snapshot string, r
 
 		Clients:         clients,
 		Namespace:       s.vsNamespace, // DataUpload is created in the same namespace as Velero server. vsNamespace is the namespace of the Velero server.
-		SANamespace:     s.vsNamespace, // The SA is created in the same namespace as Velero server. vsNamespace is the namespace of Velero	server.
+		SANamespace:     saNamespace,
 		SAName:          s.SAName,
 		TokenExpirySecs: iterator.DefaultTokenExpirySeconds,
 		MaxResults:      0, // If 0 then the CSI driver decides the value.

pkg/cmd/cli/datamover/backup.go

@@ -331,6 +331,7 @@ func (s *dataMoverBackup) createDataPathService() (dataPathService, error) {
 		s.config.changeID,
 		s.config.volumeID,
 		s.config.snapshotID,
+		s.cbtService,
 		s.logger,
 	), nil
 }

pkg/datamover/backup_micro_service.go

@@ -34,6 +34,7 @@ import (
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
@@ -71,6 +72,7 @@ type BackupMicroService struct {
 	changeID   string
 	volumeID   string
 	snapshotID string
+	cbtService cbtservice.Service
 }
 
 type dataPathResult struct {
@@ -80,7 +82,7 @@ type dataPathResult struct {
 
 func NewBackupMicroService(ctx context.Context, client client.Client, kubeClient kubernetes.Interface, dataUploadName string, namespace string, nodeName string,
 	sourceTargetPath datapath.AccessPoint, dataPathMgr *datapath.Manager, repoEnsurer *repository.Ensurer, cred *credentials.CredentialGetter,
-	duInformer cache.Informer, changeID string, volumeID string, snapshotID string, log logrus.FieldLogger) *BackupMicroService {
+	duInformer cache.Informer, changeID string, volumeID string, snapshotID string, cbtService cbtservice.Service, log logrus.FieldLogger) *BackupMicroService {
 	return &BackupMicroService{
 		ctx:              ctx,
 		client:           client,
@@ -98,6 +100,7 @@ func NewBackupMicroService(ctx context.Context, client client.Client, kubeClient
 		changeID:         changeID,
 		volumeID:         volumeID,
 		snapshotID:       snapshotID,
+		cbtService:       cbtService,
 	}
 }
 
@@ -210,6 +213,7 @@ func (r *BackupMicroService) RunCancelableDataPath(ctx context.Context) (string,
 		VolumeID:       r.volumeID,
 		ChangeID:       r.changeID,
 		SnapshotID:     r.snapshotID,
+		CBTService:     r.cbtService,
 	}); err != nil {
 		return "", errors.Wrap(err, "error starting data path backup")
 	}

pkg/datapath/data_path.go

@@ -57,6 +57,7 @@ type BackupStartParam struct {
 	VolumeID       string
 	ChangeID       string
 	SnapshotID     string
+	CBTService     cbtservice.Service
 }
 
 // RestoreStartParam define the input param for restore start
@@ -203,6 +204,7 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st
 					VolumeID: backupParam.VolumeID,
 					ChangeID: backupParam.ChangeID,
 				},
+				Service: backupParam.CBTService,
 			},
 			source.VolMode,
 			uploaderConfig,