feat: support full backup

ref: longhorn/longhorn 7070

Signed-off-by: Jack Lin <[email protected]>

backupstore.go

@@ -37,15 +37,18 @@ type ProcessingBlocks struct {
 
 type Backup struct {
  sync.Mutex
- Name string
- VolumeName string
- SnapshotName string
- SnapshotCreatedAt string
- CreatedTime string
- Size int64 `json:",string"`
- Labels map[string]string
- IsIncremental bool
- CompressionMethod string
+ Name string
+ VolumeName string
+ SnapshotName string
+ SnapshotCreatedAt string
+ CreatedTime string
+ Size int64 `json:",string"`
+ Labels map[string]string
+ Parameters map[string]string
+ IsIncremental bool
+ CompressionMethod string
+ NewlyUploadedDataSize int64 `json:",string"`
+ ReUploadedDataSize int64 `json:",string"`
 
  ProcessingBlocks *ProcessingBlocks
 

deltablock.go

@@ -16,6 +16,7 @@ import (
  . "github.com/longhorn/backupstore/logging"
  "github.com/longhorn/backupstore/types"
  "github.com/longhorn/backupstore/util"
+ lhbackup "github.com/longhorn/go-common-libs/backup"
 )
 
 type DeltaBackupConfig struct {
@@ -26,6 +27,7 @@ type DeltaBackupConfig struct {
  DeltaOps DeltaBlockBackupOperations
  Labels map[string]string
  ConcurrentLimit int32
+ Parameters map[string]string
 }
 
 type DeltaRestoreConfig struct {
@@ -181,7 +183,7 @@ func CreateDeltaBlockBackup(backupName string, config *DeltaBackupConfig) (isInc
  }
 
  backupRequest := &backupRequest{}
- if volume.LastBackupName != "" {
+ if volume.LastBackupName != "" && !isFullBackup(config) {
  lastBackupName := volume.LastBackupName
  var backup, err = loadBackup(bsDriver, lastBackupName, volume.Name)
  if err != nil {
@@ -401,21 +403,57 @@ func backupBlock(bsDriver BackupStoreDriver, config *DeltaBackupConfig,
  }()
 
  blkFile := getBlockFilePath(volume.Name, checksum)
+ reUpload := false
  if bsDriver.FileExists(blkFile) {
- log.Debugf("Found existing block matching at %v", blkFile)
- return nil
+ if !isFullBackup(config) {
+ log.Debugf("Found existing block matching at %v", blkFile)
+ return nil
+ }
+ log.Debugf("Reupload existing block matching at %v", blkFile)
+ reUpload = true
  }
 
- log.Tracef("Creating new block file at %v", blkFile)
- newBlock = true
+ log.Tracef("Uploading block file at %v", blkFile)
+ newBlock = !reUpload
  rs, err := util.CompressData(deltaBackup.CompressionMethod, block)
  if err != nil {
  return err
  }
 
+ dataSize, err := getTransferDataSize(rs)
+ if err != nil {
+ return errors.Wrapf(err, "failed to get transfer data size during saving blocks")
+ }
+ updateUploadDataSize(reUpload, deltaBackup, dataSize)
+
  return bsDriver.Write(blkFile, rs)
 }
 
+func getTransferDataSize(rs io.ReadSeeker) (int64, error) {
+ size, err := rs.Seek(0, io.SeekEnd)
+ if err != nil {
+ return 0, err
+ }
+
+ // reset to start
+ if _, err = rs.Seek(0, io.SeekStart); err != nil {
+ return 0, err
+ }
+
+ return size, nil
+}
+
+func updateUploadDataSize(reUpload bool, deltaBackup *Backup, dataSize int64) {
+ deltaBackup.Lock()
+ defer deltaBackup.Unlock()
+
+ if reUpload {
+ deltaBackup.ReUploadedDataSize += dataSize
+ } else {
+ deltaBackup.NewlyUploadedDataSize += dataSize
+ }
+}
+
 func backupMapping(bsDriver BackupStoreDriver, config *DeltaBackupConfig,
  deltaBackup *Backup, blockSize int64, mapping types.Mapping, progress *progress) error {
  volume := config.Volume
@@ -565,7 +603,10 @@ func performBackup(bsDriver BackupStoreDriver, config *DeltaBackupConfig, delta
  backup.CreatedTime = util.Now()
  backup.Size = int64(len(backup.Blocks)) * DEFAULT_BLOCK_SIZE
  backup.Labels = config.Labels
+ backup.Parameters = config.Parameters
  backup.IsIncremental = lastBackup != nil
+ backup.NewlyUploadedDataSize = deltaBackup.NewlyUploadedDataSize
+ backup.ReUploadedDataSize = deltaBackup.ReUploadedDataSize
 
  if err := saveBackup(bsDriver, backup); err != nil {
  return progress.progress, "", err
@@ -1354,3 +1395,12 @@ func getBlockNamesForVolume(driver BackupStoreDriver, volumeName string) ([]stri
 
  return util.ExtractNames(names, "", BLK_SUFFIX), nil
 }
+
+func isFullBackup(config *DeltaBackupConfig) bool {
+ if config.Parameters != nil {
+ if backupMode, exist := config.Parameters[lhbackup.LonghornBackupParameterBackupMode]; exist {
+ return lhbackup.LonghornBackupMode(backupMode) == lhbackup.LonghornBackupModeFull
+ }
+ }
+ return false
+}

go.mod

@@ -6,20 +6,21 @@ require (
  github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1
  github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0
  github.com/aws/aws-sdk-go v1.34.2
- github.com/gammazero/workerpool v1.1.3
- github.com/google/uuid v1.3.0
- github.com/pierrec/lz4/v4 v4.1.21
+ github.com/gammazero/workerpool v1.1.2
+ github.com/google/uuid v1.6.0
+ github.com/longhorn/go-common-libs v0.0.0-20240411093823-b8862efb8e03
+ github.com/pierrec/lz4/v4 v4.1.17
  github.com/pkg/errors v0.9.1
- github.com/sirupsen/logrus v1.6.0
+ github.com/sirupsen/logrus v1.9.3
  github.com/slok/goresilience v0.2.0
  github.com/spf13/afero v1.5.1
- github.com/stretchr/testify v1.8.0
+ github.com/stretchr/testify v1.8.4
  github.com/urfave/cli v1.20.0
  golang.org/x/net v0.23.0
- golang.org/x/sys v0.18.0
+ golang.org/x/sys v0.19.0
  gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c
  k8s.io/apimachinery v0.26.0
- k8s.io/mount-utils v0.26.0
+ k8s.io/mount-utils v0.29.3
 )
 
 require (
@@ -28,11 +29,10 @@ require (
  github.com/cespare/xxhash/v2 v2.1.1 // indirect
  github.com/davecgh/go-spew v1.1.1 // indirect
  github.com/gammazero/deque v0.2.0 // indirect
- github.com/go-logr/logr v1.2.3 // indirect
- github.com/golang/protobuf v1.5.2 // indirect
+ github.com/go-logr/logr v1.3.0 // indirect
+ github.com/golang/protobuf v1.5.4 // indirect
  github.com/jmespath/go-jmespath v0.3.0 // indirect
- github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
- github.com/kr/pretty v0.2.1 // indirect
+ github.com/kr/pretty v0.3.1 // indirect
  github.com/kr/text v0.2.0 // indirect
  github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
  github.com/moby/sys/mountinfo v0.6.2 // indirect
@@ -41,9 +41,10 @@ require (
  github.com/prometheus/client_model v0.2.0 // indirect
  github.com/prometheus/common v0.26.0 // indirect
  github.com/prometheus/procfs v0.6.0 // indirect
+ github.com/rogpeppe/go-internal v1.10.0 // indirect
  golang.org/x/text v0.14.0 // indirect
- google.golang.org/protobuf v1.28.1 // indirect
+ google.golang.org/protobuf v1.33.0 // indirect
  gopkg.in/yaml.v3 v3.0.1 // indirect
- k8s.io/klog/v2 v2.80.1 // indirect
- k8s.io/utils v0.0.0-20221107191617-1a15be271d1d // indirect
+ k8s.io/klog/v2 v2.110.1 // indirect
+ k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 )