fix(restore): adding support to restore in an encrypted pool (#292)

Encrypted pool does not allow the volume to be pre created for the
restore purpose. Here changing the design to do the restore first
and then create the ZFSVolume object which will bind the volume
already created while doing restore.


Signed-off-by: Pawan <pawan@mayadata.io>

pkg/apis/openebs.io/zfs/v1/zfsrestore.go

@@ -29,6 +29,7 @@ type ZFSRestore struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"` // set name to restore name + volume name + something like csp tag
 	Spec              ZFSRestoreSpec              `json:"spec"`
+	VolSpec           VolumeInfo                  `json:"volSpec,omitempty"`
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:Enum=Init;Done;Failed;Pending;InProgress;Invalid
 	Status ZFSRestoreStatus `json:"status"`

pkg/apis/openebs.io/zfs/v1/zz_generated.deepcopy.go

@@ -154,6 +154,7 @@ func (in *ZFSRestore) DeepCopyInto(out *ZFSRestore) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	out.Spec = in.Spec
+	out.VolSpec = in.VolSpec
 	return
 }
 

pkg/builder/restorebuilder/build.go

@@ -99,6 +99,12 @@ func (b *Builder) WithVolume(name string) *Builder {
 	return b
 }
 
+// WithVolSpec copies volume spec to ZFSRestore Object
+func (b *Builder) WithVolSpec(vspec apis.VolumeInfo) *Builder {
+	b.rstr.Object.VolSpec = vspec
+	return b
+}
+
 // WithNode sets the node id for ZFSRestore
 func (b *Builder) WithNode(node string) *Builder {
 	if node == "" {

pkg/zfs/zfs_util.go

@@ -21,6 +21,8 @@ import (
 	"path/filepath"
 
 	"fmt"
+	"os"
+	"time"
 
 	"strings"
 
@@ -329,11 +331,12 @@ func buildVolumeBackupArgs(bkp *apis.ZFSBackup, vol *apis.ZFSVolume) ([]string,
 }
 
 // builldVolumeRestoreArgs returns volume recv command for receiving the zfs volume
-func buildVolumeRestoreArgs(rstr *apis.ZFSRestore, vol *apis.ZFSVolume) ([]string, error) {
+func buildVolumeRestoreArgs(rstr *apis.ZFSRestore) ([]string, error) {
 	var ZFSVolArg []string
+	var ZFSRecvParam string
 	restoreSrc := rstr.Spec.RestoreSrc
 
-	volume := vol.Spec.PoolName + "/" + vol.Name
+	volume := rstr.VolSpec.PoolName + "/" + rstr.Spec.VolumeName
 
 	rstrAddr := strings.Split(restoreSrc, ":")
 	if len(rstrAddr) != 2 {
@@ -342,7 +345,36 @@ func buildVolumeRestoreArgs(rstr *apis.ZFSRestore, vol *apis.ZFSVolume) ([]strin
 
 	source := "nc -w 3 " + rstrAddr[0] + " " + rstrAddr[1] + " | "
 
-	cmd := source + ZFSVolCmd + " " + ZFSRecvArg + " -F " + volume
+	if rstr.VolSpec.VolumeType == VolTypeDataset {
+		if len(rstr.VolSpec.Capacity) != 0 {
+			ZFSRecvParam += " -o quota=" + rstr.VolSpec.Capacity
+		}
+		if len(rstr.VolSpec.RecordSize) != 0 {
+			ZFSRecvParam += " -o recordsize=" + rstr.VolSpec.RecordSize
+		}
+		if rstr.VolSpec.ThinProvision == "no" {
+			ZFSRecvParam += " -o reservation=" + rstr.VolSpec.Capacity
+		}
+		ZFSRecvParam += " -o mountpoint=legacy"
+	}
+
+	if len(rstr.VolSpec.Dedup) != 0 {
+		ZFSRecvParam += " -o dedup=" + rstr.VolSpec.Dedup
+	}
+	if len(rstr.VolSpec.Compression) != 0 {
+		ZFSRecvParam += " -o compression=" + rstr.VolSpec.Compression
+	}
+	if len(rstr.VolSpec.Encryption) != 0 {
+		ZFSRecvParam += " -o encryption=" + rstr.VolSpec.Encryption
+	}
+	if len(rstr.VolSpec.KeyLocation) != 0 {
+		ZFSRecvParam += " -o keylocation=" + rstr.VolSpec.KeyLocation
+	}
+	if len(rstr.VolSpec.KeyFormat) != 0 {
+		ZFSRecvParam += " -o keyformat=" + rstr.VolSpec.KeyFormat
+	}
+
+	cmd := source + ZFSVolCmd + " " + ZFSRecvArg + ZFSRecvParam + " -F " + volume
 
 	ZFSVolArg = append(ZFSVolArg, "-c", cmd)
 
@@ -794,18 +826,43 @@ func DestoryBackup(bkp *apis.ZFSBackup) error {
 	return err
 }
 
+// getDevice waits for the device to be created and returns the devpath
+func getDevice(volume string) (string, error) {
+	device := ZFSDevPath + volume
+	// device should be created within 5 seconds
+	timeout := time.After(5 * time.Second)
+	for {
+		select {
+		case <-timeout:
+			return "", fmt.Errorf("zfs: not able to get the device: %s", device)
+		default:
+			if _, err := os.Stat(device); err == nil {
+				return device, nil
+			}
+		}
+		time.Sleep(1 * time.Second)
+	}
+}
+
 // CreateRestore creates the restore
 func CreateRestore(rstr *apis.ZFSRestore) error {
-	vol, err := GetZFSVolume(rstr.Spec.VolumeName)
-	if err != nil {
-		return err
-	}
-	volume := vol.Spec.PoolName + "/" + vol.Name
-	args, err := buildVolumeRestoreArgs(rstr, vol)
+	if len(rstr.VolSpec.PoolName) == 0 {
+		// for backward compatibility, older version of
+		// velero will not add spec in the ZFSRestore Object
+		// query it here and fill that information
+		vol, err := GetZFSVolume(rstr.Spec.VolumeName)
+		if err != nil {
+			return err
+		}
+		rstr.VolSpec = vol.Spec
+	}
+	args, err := buildVolumeRestoreArgs(rstr)
 	if err != nil {
 		return err
 	}
 
+	volume := rstr.VolSpec.PoolName + "/" + rstr.Spec.VolumeName
+
 	cmd := exec.Command("bash", args...)
 	out, err := cmd.CombinedOutput()
 
@@ -820,12 +877,18 @@ func CreateRestore(rstr *apis.ZFSRestore) error {
 	 * need to generate a new uuid for zfs and btrfs volumes
 	 * so that we can mount it.
 	 */
-	if vol.Spec.FsType == "xfs" {
-		device := ZFSDevPath + volume
+	if rstr.VolSpec.FsType == "xfs" {
+		device, err := getDevice(volume)
+		if err != nil {
+			return err
+		}
 		return xfs.GenerateUUID(device)
 	}
-	if vol.Spec.FsType == "btrfs" {
-		device := ZFSDevPath + volume
+	if rstr.VolSpec.FsType == "btrfs" {
+		device, err := getDevice(volume)
+		if err != nil {
+			return err
+		}
 		return btrfs.GenerateUUID(device)
 	}