Merge branch 'main' into dependabot/github_actions/docker/setup-buildx-action-4

.github/workflows/nightly-trivy-scan.yml

@@ -31,6 +31,6 @@ jobs:
           output: 'trivy-results.sarif'
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v3
+        uses: github/codeql-action/upload-sarif@v4.37.3
         with:
           sarif_file: 'trivy-results.sarif'

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.
@@ -110,6 +116,12 @@ func (s *ServiceImpl) GetChangedBlocks(ctx context.Context, snapshot string, cha
 		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,
 		PrevSnapshotID: changeID,
@@ -120,7 +132,7 @@ func (s *ServiceImpl) GetChangedBlocks(ctx context.Context, snapshot string, cha
 
 		Clients:         clients,
 		Namespace:       s.vsNamespace,
-		SANamespace:     s.vsNamespace,
+		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

@@ -23,23 +23,23 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
 	cachetool "k8s.io/client-go/tools/cache"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
+	veleroshared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	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"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
-
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
 const (
@@ -71,6 +71,7 @@ type BackupMicroService struct {
 	changeID   string
 	volumeID   string
 	snapshotID string
+	cbtService cbtservice.Service
 }
 
 type dataPathResult struct {
@@ -80,7 +81,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 +99,7 @@ func NewBackupMicroService(ctx context.Context, client client.Client, kubeClient
 		changeID:         changeID,
 		volumeID:         volumeID,
 		snapshotID:       snapshotID,
+		cbtService:       cbtService,
 	}
 }
 
@@ -202,14 +204,23 @@ func (r *BackupMicroService) RunCancelableDataPath(ctx context.Context) (string,
 		velerov1api.AsyncOperationIDLabel: du.Labels[velerov1api.AsyncOperationIDLabel],
 	}
 
+	// Modify the ParentSnapshot to "" and ForceFull to true when ParentSnapshot is "none".
+	parentSnapshot := du.Spec.ParentSnapshot
+	forceFull := false
+	if du.Spec.ParentSnapshot == veleroshared.DataUploadParentSnapshotNone {
+		parentSnapshot = ""
+		forceFull = true
+	}
+
 	if err := dp.StartBackup(r.sourceTargetPath, du.Spec.DataMoverConfig, &datapath.BackupStartParam{
 		RealSource:     GetRealSource(du.Spec.SourceNamespace, du.Spec.SourcePVC),
-		ParentSnapshot: du.Spec.ParentSnapshot,
-		ForceFull:      false,
+		ParentSnapshot: parentSnapshot,
+		ForceFull:      forceFull,
 		Tags:           tags,
 		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,

pkg/uploader/block/uploader.go

@@ -26,6 +26,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
@@ -62,10 +63,11 @@ type Uploader interface {
 }
 
 type blockUploader struct {
-	ctx        context.Context
-	repoWriter udmrepo.BackupRepo
-	progress   uploader.ProgressUpdater
-	log        logrus.FieldLogger
+	ctx                context.Context
+	repoWriter         udmrepo.BackupRepo
+	progress           uploader.ProgressUpdater
+	log                logrus.FieldLogger
+	lastProgressUpdate time.Time
 }
 
 func NewUploader(ctx context.Context, repoWriter udmrepo.BackupRepo, progress uploader.ProgressUpdater, log logrus.FieldLogger) Uploader {
@@ -198,6 +200,17 @@ func (blkup *blockUploader) backupObject(dev *os.File, dest udmrepo.ObjectWriter
 	return id, backupSize, objectSize, err
 }
 
+func (blkup *blockUploader) UpdateProgress(p *uploader.Progress) {
+	if blkup.progress == nil {
+		return
+	}
+
+	if time.Since(blkup.lastProgressUpdate) >= 10*time.Second || p.BytesDone == p.TotalBytes {
+		blkup.progress.UpdateProgress(p)
+		blkup.lastProgressUpdate = time.Now()
+	}
+}
+
 type readResult struct {
 	buffer []byte
 	offset int64
@@ -233,7 +246,7 @@ func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.Object
 	go func() {
 		defer wg.Done()
 		defer close(quit)
-		written, lastPos, writeErr = backupWriteProc(blkup.ctx, writer, resultChan, list, aligned, totalCount, int(blockSize), blkup.progress)
+		written, lastPos, writeErr = backupWriteProc(blkup.ctx, writer, resultChan, list, aligned, totalCount, int(blockSize), blkup)
 	}()
 
 	wg.Wait()
@@ -250,7 +263,7 @@ func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.Object
 
 		written += s
 
-		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: aligned, TotalBytes: aligned})
+		blkup.UpdateProgress(&uploader.Progress{BytesDone: aligned, TotalBytes: aligned})
 	}
 
 	return written, aligned, nil
@@ -419,7 +432,7 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 	go func() {
 		defer wg.Done()
 		defer close(quit)
-		written, writeErr = restoreWriteProc(blkup.ctx, dest, resultChan, list, totalLength, totalCount, int(blockSize), destPath, blkup.progress, blkup.log)
+		written, writeErr = restoreWriteProc(blkup.ctx, dest, resultChan, list, totalLength, totalCount, int(blockSize), destPath, blkup, blkup.log)
 	}()
 
 	wg.Wait()