feat: support full backup

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,19 +403,60 @@ 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
 	}
 
-	return bsDriver.Write(blkFile, rs)
+	dataSize, err := getTransferDataSize(rs)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get transfer data size during saving blocks")
+	}
+
+	if err := bsDriver.Write(blkFile, rs); err != nil {
+		return errors.Wrapf(err, "failed to write data during saving blocks")
+	}
+
+	updateUploadDataSize(reUpload, deltaBackup, dataSize)
+
+	return nil
+}
+
+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,
@@ -565,7 +608,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 +1400,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
 )