diff --git a/.github/workflows/nightly-trivy-scan.yml b/.github/workflows/nightly-trivy-scan.yml index be0aa4dcf..fc63b27d2 100644 --- a/.github/workflows/nightly-trivy-scan.yml +++ b/.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' \ No newline at end of file diff --git a/pkg/cbtservice/csi_service_impl.go b/pkg/cbtservice/csi_service_impl.go index 4d0ea3fca..235477bd4 100644 --- a/pkg/cbtservice/csi_service_impl.go +++ b/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. diff --git a/pkg/cmd/cli/datamover/backup.go b/pkg/cmd/cli/datamover/backup.go index 07ac7dc18..aa0b2bcfb 100644 --- a/pkg/cmd/cli/datamover/backup.go +++ b/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 } diff --git a/pkg/datamover/backup_micro_service.go b/pkg/datamover/backup_micro_service.go index cb5aeb3fe..81912f600 100644 --- a/pkg/datamover/backup_micro_service.go +++ b/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") } diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index 6e36ce6af..2ec750805 100644 --- a/pkg/datapath/data_path.go +++ b/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, diff --git a/pkg/uploader/block/uploader.go b/pkg/uploader/block/uploader.go index 2a1446b83..1d74bd462 100644 --- a/pkg/uploader/block/uploader.go +++ b/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()