feat(zfspv): adding support to do incremental backup/restore (#121)

We can create the snapshot location as below

```yaml
apiVersion: velero.io/v1
kind: VolumeSnapshotLocation
metadata:
  name: incr
  namespace: velero
spec:
  provider: openebs.io/zfspv-blockstore
  config:
    bucket: velero
    prefix: zfs
    incrBackupCount: "3" # number of incremental backup we want to have
    namespace: openebs # this is namespace where ZFS-LocalPV creates all the CRs, passed as OPENEBS_NAMESPACE env in the ZFS-LocalPV deployment
    provider: aws
    region: minio
    s3ForcePathStyle: "true"
    s3Url: http://minio.velero.svc:9000
```

here we can specify how many incremental backups we want to have. We will create a full backup first and if we have provided non zero value for `incrBackupCount` option, then the plugin will create that many incremental backup. Thing to note here is `incrBackupCount` parameter defines how many incremental backup we want, it does not include the first full backup.

The plugin will create the incremental backup if it is scheduled backup. If it is not a scheduled backup, the ZFS-LocalPV plugin will create full backup for each request.

While doing the restore, we just need to give the backup name which we want to restore. The plugin is capable of identifying the full backup of the incremental snapshot group and will restore from the full backup and keep restoring the incremental backup till the backup name provided in the restore command.

Note: We should not modify the VolumeSnapshotLocation once it is created.

Signed-off-by: Pawan <[email protected]>

changelogs/unreleased/121-pawanpraka1

@@ -0,0 +1 @@
+adding support to do incremental backup/restore for ZFS-LocalPV

pkg/clouduploader/operation.go

@@ -16,13 +16,34 @@ limitations under the License.
 
 package clouduploader
 
-import "github.com/aws/aws-sdk-go/service/s3/s3manager"
+import (
+	"io"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	"gocloud.dev/blob"
+)
 
 const (
 	// backupDir is remote storage-bucket directory
 	backupDir = "backups"
 )
 
+const (
+	// Type of Key, used while listing keys
+
+	// KeyFile - if key is a file
+	ListKeyFile int = 1 << iota
+
+	// KeyDirectory - if key is a directory
+	ListKeyDir
+
+	// KeyBoth - if key is a file or directory
+	ListKeyBoth
+)
+
 // Upload will perform upload operation for given file.
 // It will create a TCP server through which client can
 // connect and upload data to cloud blob storage file
@@ -146,3 +167,90 @@ func (c *Conn) ConnReadyWait() bool {
 	ok := <-*c.ConnReady
 	return ok
 }
+
+// listKeys return list of Keys -- files/directories
+// Note:
+// - list may contain incomplete list of keys, check for error before using list
+// - listKeys uses '/' as delimiter.
+func (c *Conn) listKeys(prefix string, keyType int) ([]string, error) {
+	keys := []string{}
+
+	lister := c.bucket.List(&blob.ListOptions{
+		Delimiter: "/",
+		Prefix:    prefix,
+	})
+	for {
+		obj, err := lister.Next(c.ctx)
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			c.Log.Errorf("Failed to get next blob err=%v", err)
+			return keys, err
+		}
+
+		switch keyType {
+		case ListKeyBoth:
+		case ListKeyFile:
+			if obj.IsDir {
+				continue
+			}
+		case ListKeyDir:
+			if !obj.IsDir {
+				continue
+			}
+		default:
+			c.Log.Warningf("Invalid keyType=%d, Ignored", keyType)
+			continue
+		}
+
+		keys = append(keys, obj.Key)
+	}
+	return keys, nil
+}
+
+// bkpPathPrefix return 'prefix path' for the given 'backup name prefix'
+func (c *Conn) bkpPathPrefix(backupPrefix string) string {
+	if c.backupPathPrefix == "" {
+		return backupDir + "/" + backupPrefix
+	}
+	return c.backupPathPrefix + "/" + backupDir + "/" + backupPrefix
+}
+
+// filePathPrefix generate prefix for the given file name prefix using 'configured file prefix'
+func (c *Conn) filePathPrefix(filePrefix string) string {
+	return c.prefix + "-" + filePrefix
+}
+
+// GetSnapListFromCloud gets the list of a snapshot for the given backup name
+// the argument should be same as that of GenerateRemoteFilename(file, backup) call
+// used while doing the backup of the volume
+func (c *Conn) GetSnapListFromCloud(file, backup string) ([]string, error) {
+	var snapList []string
+
+	// list directory having schedule/backup name as prefix
+	dirs, err := c.listKeys(c.bkpPathPrefix(backup), ListKeyDir)
+	if err != nil {
+		return snapList, errors.Wrapf(err, "failed to get list of directory")
+	}
+
+	for _, dir := range dirs {
+		// list files for dir having volume name as prefix
+		files, err := c.listKeys(dir+c.filePathPrefix(file), ListKeyFile)
+		if err != nil {
+			return snapList, errors.Wrapf(err, "failed to get list of snapshot file at path=%v", dir)
+		}
+
+		if len(files) != 0 {
+			// snapshot exist in the backup directory
+
+			// add backup name from dir path to snapList
+			s := strings.Split(dir, "/")
+
+			// dir will contain path with trailing '/', example: 'backups/b-0/'
+			snapList = append(snapList, s[len(s)-2])
+		}
+	}
+	return snapList, nil
+}

pkg/zfs/plugin/backup.go

@@ -83,6 +83,14 @@ func (p *Plugin) getPrevSnap(volname, schdname string) (string, error) {
 
 	var backups []string
 
+	size := len(bkpList.Items)
+	count := p.incremental + 1
+
+	if uint64(size)%count == 0 {
+		// have to start the next snapshot incremental group, take the full backup
+		return "", nil
+	}
+
 	/*
 	 * Backup names are in the form of <schdeule>-<yyyymmddhhmmss>
 	 * to get the last snapshot, sort the list of successful backups,
@@ -115,15 +123,15 @@ func (p *Plugin) createBackup(vol *apis.ZFSVolume, schdname, snapname string, po
 		// add schdeule name as label
 		labels[VeleroSchdKey] = schdname
 		labels[VeleroVolKey] = vol.Name
-		if p.incremental {
-			prevSnap, err = p.getPrevSnap(vol.Name, schdname)
-			if err != nil {
-				p.Log.Errorf("zfs: Failed to get prev snapshot bkp %s err: {%v}", snapname, err)
-				return "", err
-			}
+		prevSnap, err = p.getPrevSnap(vol.Name, schdname)
+		if err != nil {
+			p.Log.Errorf("zfs: Failed to get prev snapshot bkp %s err: {%v}", snapname, err)
+			return "", err
 		}
 	}
 
+	p.Log.Debugf("zfs: backup incr(%d) schd=%s snap=%s prevsnap=%s vol=%s", p.incremental, schdname, snapname, prevSnap, vol.Name)
+
 	serverAddr := p.remoteAddr + ":" + strconv.Itoa(port)
 
 	bkp, err := bkpbuilder.NewBuilder().