feat(ZFSPV): adding support for applications to create "zfs" flesystem (#15)

Application can now create a storageclass to create zfs filesystem apiVersion: storage.k8s.io/v1 kind: StorageClass metadata: name: openebs-zfspv5 allowVolumeExpansion: true parameters: blocksize: "4k" fstype: "zfs" poolname: "zfspv-pool" provisioner: zfs.csi.openebs.io ZFSPV was supporting ext2/3/4 and xfs filesystem only which adds one extra filesystem layer on top of ZFS filesystem. So now we can driectly write to the ZFS filesystem and get the optimal performance by directly creating ZFS filesystem for storage. Signed-off-by: Pawan <pawan@mayadata.io>
2025-12-11 22:10:11 +01:00 · 2019-11-21 19:00:15 +05:30 · 2019-11-21 19:00:15 +05:30 · 68db6d2774
commit 68db6d2774
parent 4ffd857191
13 changed files with 428 additions and 176 deletions
--- a/cmd/controller/controller.go
+++ b/cmd/controller/controller.go
@ -80,7 +80,7 @@ func (c *ZVController) syncZV(zv *apis.ZFSVolume) error {
 	var err error
 	// ZFS Volume should be deleted. Check if deletion timestamp is set
 	if c.isDeletionCandidate(zv) {
-		err = zvol.DestroyZvol(zv)
+		err = zvol.DestroyVolume(zv)
 		if err == nil {
 			zvol.RemoveZvolFinalizer(zv)
 		}
@ -91,7 +91,7 @@ func (c *ZVController) syncZV(zv *apis.ZFSVolume) error {
 		if zv.Finalizers != nil {
 			err = zvol.SetZvolProp(zv)
 		} else {
-			err = zvol.CreateZvol(zv)
+			err = zvol.CreateVolume(zv)
 			if err == nil {
 				err = zvol.UpdateZvolInfo(zv)
 			}
--- a/deploy/sample/fio.yaml
+++ b/deploy/sample/fio.yaml
@ -4,13 +4,14 @@ metadata:
  name: openebs-zfspv
 allowVolumeExpansion: true
 parameters:
-  blocksize: "4k"
+  recordsize: "4k"
  compression: "on"
  dedup: "on"
  thinprovision: "yes"
  #encryption: "on"
  #keyformat: "raw"
  #keylocation: "file:///home/pawan/key"
+  fstype: "zfs"
  poolname: "zfspv-pool"
 provisioner: zfs.csi.openebs.io
 allowedTopologies:
@ -32,22 +33,33 @@ spec:
    requests:
      storage: 4Gi
 ---
-apiVersion: v1
-kind: Pod
+apiVersion: apps/v1
+kind: Deployment
 metadata:
  name: fio
+  labels:
+    name: fio
 spec:
-  restartPolicy: Never
-  containers:
-  - name: perfrunner
-    image: openebs/tests-fio
-    command: ["/bin/bash"]
-    args: ["-c", "while true ;do sleep 50; done"]
-    volumeMounts:
-       - mountPath: /datadir
-         name: fio-vol
-    tty: true
-  volumes:
-  - name: fio-vol
-    persistentVolumeClaim:
-      claimName: csi-zfspv
+  replicas: 1
+  selector:
+    matchLabels:
+      name: fio
+  template:
+    metadata:
+      labels:
+        name: fio
+    spec:
+      containers:
+        - resources:
+          name: perfrunner
+          image: openebs/tests-fio
+          imagePullPolicy: IfNotPresent
+          command: ["/bin/bash"]
+          args: ["-c", "while true ;do sleep 50; done"]
+          volumeMounts:
+            - mountPath: /datadir
+              name: fio-vol
+      volumes:
+        - name: fio-vol
+          persistentVolumeClaim:
+            claimName: csi-zfspv
--- a/deploy/sample/mongo-statefulset.yaml
+++ b/deploy/sample/mongo-statefulset.yaml
@ -7,9 +7,9 @@ kind: StorageClass
 metadata:
  name: mongo-pv-az
 parameters:
-  blocksize: "4k"
+  volblocksize: "4k"
  poolname: "zfspv-pool"
-  fsType: "xfs"
+  fstype: "xfs"
 provisioner: zfs.csi.openebs.io
 ---
 # Headless service for stable DNS entries of StatefulSet members.
--- a/deploy/sample/percona.yaml
+++ b/deploy/sample/percona.yaml
@ -4,7 +4,7 @@ metadata:
  name: openebs-zfspv
 allowVolumeExpansion: true
 parameters:
-  blocksize: "4k"
+  volblocksize: "4k"
  compression: "on"
  dedup: "on"
  thinprovision: "yes"
--- a/deploy/sample/zfspvcr.yaml
+++ b/deploy/sample/zfspvcr.yaml
@ -4,7 +4,6 @@ metadata:
  name: pvc-37b07ad6-db68-11e9-bbb6-000c296e38d9
  namespace: openebs
 spec:
-  blocksize: 4k
  capacity: "4294967296"
  compression: "off"
  dedup: "off"
@ -13,5 +12,6 @@ spec:
  keylocation: ""
  ownerNodeID: zfspv-node1
  poolName: zfspv-pool
+  recordsize: 8k
  thinProvision: "off"
-
+  volumeType: DATASET
--- a/deploy/zfs-operator.yaml
+++ b/deploy/zfs-operator.yaml
@ -20,7 +20,7 @@ spec:
    singular: zfsvolume
    kind: ZFSVolume
    shortNames:
-    - zvol
+    - zfsvol
    - zv
  additionalPrinterColumns:
  - JSONPath: .spec.poolName
@ -35,6 +35,18 @@ spec:
    name: Size
    description: Size of the volume
    type: string
+  - JSONPath: .spec.volblocksize
+    name: volblocksize
+    description: volblocksize for the created zvol
+    type: string
+  - JSONPath: .spec.recordsize
+    name: recordsize
+    description: recordsize for the created zfs dataset
+    type: string
+  - JSONPath: .spec.fsType
+    name: Filesystem
+    description: filesystem created on the volume
+    type: string

 ---
 ##############################################
--- a/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go
+++ b/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go
@ -82,31 +82,42 @@ type VolumeInfo struct {
 	// Capacity of the volume
 	Capacity string `json:"capacity"`

-	// BlockSize specifies the blocksize
-	// which we should use to create the zvol
-	BlockSize string `json:"blocksize"`
+	// RecordSize specifies the record size
+	// for the zfs dataset
+	RecordSize string `json:"recordsize,omitempty"`
+
+	// VolBlockSize specifies the block size for the zvol
+	VolBlockSize string `json:"volblocksize,omitempty"`

 	// Compression specifies if the it should
 	// enabled on the zvol
-	Compression string `json:"compression"`
+	Compression string `json:"compression,omitempty"`

 	// Dedup specifies the deduplication
 	// should be enabled on the zvol
-	Dedup string `json:"dedup"`
+	Dedup string `json:"dedup,omitempty"`

 	// Encryption specifies the encryption
 	// should be enabled on the zvol
-	Encryption string `json:"encryption"`
+	Encryption string `json:"encryption,omitempty"`

 	// KeyLocation is the location of key
 	// for the encryption
-	KeyLocation string `json:"keylocation"`
+	KeyLocation string `json:"keylocation,omitempty"`

 	// KeyFormat specifies format of the
 	// encryption key
-	KeyFormat string `json:"keyformat"`
+	KeyFormat string `json:"keyformat,omitempty"`

 	// Thinprovision specifies if we should
 	// thin provisioned the volume or not
-	ThinProvision string `json:"thinProvision"`
+	ThinProvision string `json:"thinProvision,omitempty"`
+
+	// VolumeType specifies whether the volume is
+	// zvol or a dataset
+	VolumeType string `json:"volumeType"`
+
+	// FsType specifies filesystem type for the
+	// zfs volume/dataset
+	FsType string `json:"fsType,omitempty"`
 }
--- a/pkg/builder/build.go
+++ b/pkg/builder/build.go
@ -142,13 +142,27 @@ func (b *Builder) WithOwnerNode(host string) *Builder {
 	return b
 }

-// WithBlockSize sets blocksize of ZFSVolume
-func (b *Builder) WithBlockSize(blockSize string) *Builder {
-	bs := "4k"
-	if len(blockSize) > 0 {
-		bs = blockSize
-	}
-	b.volume.Object.Spec.BlockSize = bs
+// WithRecordSize sets the recordsize of ZFSVolume
+func (b *Builder) WithRecordSize(rs string) *Builder {
+	b.volume.Object.Spec.RecordSize = rs
+	return b
+}
+
+// WithVolBlockSize sets the volblocksize of ZFSVolume
+func (b *Builder) WithVolBlockSize(bs string) *Builder {
+	b.volume.Object.Spec.VolBlockSize = bs
+	return b
+}
+
+// WithVolumeType sets if ZFSVolume needs to be thin provisioned
+func (b *Builder) WithVolumeType(vtype string) *Builder {
+	b.volume.Object.Spec.VolumeType = vtype
+	return b
+}
+
+// WithFsType sets filesystem for the ZFSVolume
+func (b *Builder) WithFsType(fstype string) *Builder {
+	b.volume.Object.Spec.FsType = fstype
 	return b
 }

--- a/pkg/driver/agent.go
+++ b/pkg/driver/agent.go
@ -22,7 +22,7 @@ import (
 	ctrl "github.com/openebs/zfs-localpv/cmd/controller"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
 	"github.com/openebs/zfs-localpv/pkg/builder"
-	zvol "github.com/openebs/zfs-localpv/pkg/zfs"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@ -65,7 +65,7 @@ func GetVolAndMountInfo(

 	getOptions := metav1.GetOptions{}
 	vol, err := builder.NewKubeclient().
-		WithNamespace(zvol.OpenEBSNamespace).
+		WithNamespace(zfs.OpenEBSNamespace).
 		Get(req.GetVolumeId(), getOptions)

 	if err != nil {
@ -96,8 +96,8 @@ func (ns *node) NodePublishVolume(
 	if err != nil {
 		goto PublishVolumeResponse
 	}
-	// Create the zfs volume and attempt mount operation on the requested path
-	if err = zvol.CreateAndMountZvol(vol, mountInfo); err != nil {
+	// attempt mount operation on the requested path
+	if err = zfs.MountVolume(vol, mountInfo); err != nil {
 		goto PublishVolumeResponse
 	}

@ -120,6 +120,7 @@ func (ns *node) NodeUnpublishVolume(
 	var (
 		err           error
 		vol           *apis.ZFSVolume
+		devpath       string
 		currentMounts []string
 	)

@ -130,22 +131,19 @@ func (ns *node) NodeUnpublishVolume(
 	targetPath := req.GetTargetPath()
 	volumeID := req.GetVolumeId()

-	getOptions := metav1.GetOptions{}
-	vol, err = builder.NewKubeclient().
-		WithNamespace(zvol.OpenEBSNamespace).
-		Get(volumeID, getOptions)
-
-	if err != nil {
+	if vol, err = zfs.GetZFSVolume(volumeID); err != nil {
 		return nil, err
 	}

-	zfsvolume := vol.Spec.PoolName + "/" + vol.Name
-	devpath := zvol.ZFS_DEVPATH + zfsvolume
-	currentMounts, err = zvol.GetMounts(devpath)
+	if devpath, err = zfs.GetVolumeDevPath(vol); err != nil {
+		goto NodeUnpublishResponse
+	}
+
+	currentMounts, err = zfs.GetMounts(devpath)
 	if err != nil {
 		return nil, err
 	} else if len(currentMounts) == 0 {
-		goto NodeUnpublishResponse
+		return nil, status.Error(codes.Internal, "umount request for not mounted volume")
 	} else if len(currentMounts) == 1 {
 		if currentMounts[0] != targetPath {
 			return nil, status.Error(codes.Internal, "device not mounted at right path")
@ -158,15 +156,14 @@ func (ns *node) NodeUnpublishVolume(
 		return nil, status.Error(codes.Internal, "device not mounted at rightpath")
 	}

-	if vol, err = zvol.GetZFSVolume(volumeID); (err != nil) || (vol == nil) {
-		goto NodeUnpublishResponse
-	}
-
-	if err = zvol.UmountVolume(vol, req.GetTargetPath()); err != nil {
+	if err = zfs.UmountVolume(vol, req.GetTargetPath()); err != nil {
 		goto NodeUnpublishResponse
 	}

 NodeUnpublishResponse:
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
 	logrus.Infof("hostpath: volume %s path: %s has been unmounted.",
 		volumeID, targetPath)

@ -181,7 +178,7 @@ func (ns *node) NodeGetInfo(
 	req *csi.NodeGetInfoRequest,
 ) (*csi.NodeGetInfoResponse, error) {

-	topology := map[string]string{zvol.ZFSTopologyKey: ns.driver.config.NodeID}
+	topology := map[string]string{zfs.ZFSTopologyKey: ns.driver.config.NodeID}
 	return &csi.NodeGetInfoResponse{
 		NodeId: ns.driver.config.NodeID,
 		AccessibleTopology: &csi.Topology{
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@ -25,7 +25,7 @@ import (
 	"github.com/openebs/zfs-localpv/pkg/builder"
 	errors "github.com/openebs/zfs-localpv/pkg/common/errors"
 	csipayload "github.com/openebs/zfs-localpv/pkg/response"
-	zvol "github.com/openebs/zfs-localpv/pkg/zfs"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@ -69,7 +69,8 @@ func (cs *controller) CreateVolume(

 	volName := req.GetName()
 	size := req.GetCapacityRange().RequiredBytes
-	bs := req.GetParameters()["blocksize"]
+	rs := req.GetParameters()["recordsize"]
+	bs := req.GetParameters()["volblocksize"]
 	compression := req.GetParameters()["compression"]
 	dedup := req.GetParameters()["dedup"]
 	encr := req.GetParameters()["encryption"]
@ -78,6 +79,9 @@ func (cs *controller) CreateVolume(
 	pool := req.GetParameters()["poolname"]
 	tp := req.GetParameters()["thinprovision"]
 	schld := req.GetParameters()["scheduler"]
+	fstype := req.GetParameters()["fstype"]
+
+	vtype := zfs.GetVolumeType(fstype)

 	selected := scheduler(req.AccessibilityRequirements, schld, pool)

@ -90,7 +94,8 @@ func (cs *controller) CreateVolume(
 	volObj, err := builder.NewBuilder().
 		WithName(volName).
 		WithCapacity(strconv.FormatInt(int64(size), 10)).
-		WithBlockSize(bs).
+		WithRecordSize(rs).
+		WithVolBlockSize(bs).
 		WithPoolName(pool).
 		WithDedup(dedup).
 		WithEncryption(encr).
@ -98,18 +103,20 @@ func (cs *controller) CreateVolume(
 		WithKeyLocation(kl).
 		WithThinProv(tp).
 		WithOwnerNode(selected).
+		WithVolumeType(vtype).
+		WithFsType(fstype).
 		WithCompression(compression).Build()

 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}

-	err = zvol.ProvisionVolume(size, volObj)
+	err = zfs.ProvisionVolume(size, volObj)
 	if err != nil {
 		return nil, status.Error(codes.Internal, "not able to provision the volume")
 	}

-	topology := map[string]string{zvol.ZFSTopologyKey: selected}
+	topology := map[string]string{zfs.ZFSTopologyKey: selected}

 	return csipayload.NewCreateVolumeResponseBuilder().
 		WithName(volName).
@ -136,13 +143,13 @@ func (cs *controller) DeleteVolume(
 	volumeID := req.GetVolumeId()

 	// verify if the volume has already been deleted
-	vol, err := zvol.GetVolume(volumeID)
+	vol, err := zfs.GetVolume(volumeID)
 	if vol != nil && vol.DeletionTimestamp != nil {
 		goto deleteResponse
 	}

 	// Delete the corresponding ZV CR
-	err = zvol.DeleteVolume(volumeID)
+	err = zfs.DeleteVolume(volumeID)
 	if err != nil {
 		return nil, errors.Wrapf(
 			err,
--- a/pkg/driver/scheduler.go
+++ b/pkg/driver/scheduler.go
@ -76,7 +76,8 @@ func volumeWeightedScheduler(topo *csi.TopologyRequirement, pool string) string
 // the given zfs pool.
 func scheduler(topo *csi.TopologyRequirement, schld string, pool string) string {

-	if len(topo.Preferred) == 0 {
+	if topo == nil ||
+		len(topo.Preferred) == 0 {
 		logrus.Errorf("topology information not provided")
 		return ""
 	}
--- a/pkg/zfs/mount.go
+++ b/pkg/zfs/mount.go
@ -3,7 +3,6 @@ package zfs
 import (
 	"fmt"
 	"os"
-	"path/filepath"

 	"github.com/Sirupsen/logrus"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
@ -25,7 +24,6 @@ func FormatAndMountZvol(devicePath string, mountInfo *apis.MountInfo) error {
 		return err
 	}

-	logrus.Infof("created zvol %v and mounted %v fs %v", devicePath, mountInfo.MountPath, mountInfo.FSType)
 	return nil
 }

@ -53,12 +51,22 @@ func UmountVolume(vol *apis.ZFSVolume, targetPath string,
 		return nil
 	}

-	if err = mounter.Unmount(targetPath); err != nil {
-		logrus.Errorf(
-			"zfspv umount volume: failed to unmount: %s\nError: %v",
-			targetPath, err,
-		)
-		return err
+	if vol.Spec.VolumeType == VOLTYPE_DATASET {
+		if err = UmountZFSDataset(vol); err != nil {
+			logrus.Errorf(
+				"zfspv failed to umount dataset: path %s Error: %v",
+				targetPath, err,
+			)
+			return err
+		}
+	} else {
+		if err = mounter.Unmount(targetPath); err != nil {
+			logrus.Errorf(
+				"zfspv failed to unmount zvol: path %s Error: %v",
+				targetPath, err,
+			)
+			return err
+		}
 	}

 	if err := os.RemoveAll(targetPath); err != nil {
@ -72,7 +80,7 @@ func UmountVolume(vol *apis.ZFSVolume, targetPath string,
 }

 // GetMounts gets mountpoints for the specified volume
-func GetMounts(devicepath string) ([]string, error) {
+func GetMounts(dev string) ([]string, error) {

 	var (
 		currentMounts []string
@ -80,10 +88,6 @@ func GetMounts(devicepath string) ([]string, error) {
 		mountList     []mount.MountPoint
 	)

-	dev, err := filepath.EvalSymlinks(devicepath)
-	if err != nil {
-		return nil, err
-	}
 	mounter := mount.New("")
 	// Get list of mounted paths present with the node
 	if mountList, err = mounter.List(); err != nil {
@ -97,10 +101,8 @@ func GetMounts(devicepath string) ([]string, error) {
 	return currentMounts, nil
 }

-// CreateAndMountZvol creates the zfs Volume
-// and mounts the disk to the specified path
-func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
-	if len(mount.MountPath) == 0 {
+func verifyMountRequest(vol *apis.ZFSVolume, mountpath string) error {
+	if len(mountpath) == 0 {
 		return status.Error(codes.InvalidArgument, "mount path missing in request")
 	}

@ -109,9 +111,11 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 		return status.Error(codes.Internal, "volume is owned by different node")
 	}

-	devicePath, err := GetDevicePath(vol)
+	devicePath, err := GetVolumeDevPath(vol)
 	if err != nil {
-		return status.Error(codes.Internal, "not able to get the device path")
+		logrus.Errorf("can not get device for volume:%s dev %s err: %v",
+			vol.Name, devicePath, err.Error())
+		return err
 	}

 	/*
@ -123,18 +127,63 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 	 */
 	currentMounts, err := GetMounts(devicePath)
 	if err != nil {
+		logrus.Errorf("can not get mounts for volume:%s dev %s err: %v",
+			vol.Name, devicePath, err.Error())
 		return err
 	} else if len(currentMounts) >= 1 {
 		logrus.Errorf(
-			"can not mount, more than one mounts for volume:%s dev %s mounts: %v",
+			"can not mount, volume:%s already mounted dev %s mounts: %v",
 			vol.Name, devicePath, currentMounts,
 		)
 		return status.Error(codes.Internal, "device already mounted")
 	}
+	return nil
+}
+
+// MountZvol mounts the disk to the specified path
+func MountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	err := verifyMountRequest(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "zvol can not be mounted")
+	}
+
+	devicePath := ZFS_DEVPATH + volume
+
 	err = FormatAndMountZvol(devicePath, mount)
 	if err != nil {
-		return status.Error(codes.Internal, "not able to mount the volume")
+		return status.Error(codes.Internal, "not able to format and mount the zvol")
 	}

+	logrus.Infof("zvol %v mounted %v fs %v", volume, mount.MountPath, mount.FSType)
+
 	return err
 }
+
+// MountDataset mounts the zfs dataset to the specified path
+func MountDataset(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	err := verifyMountRequest(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "dataset can not be mounted")
+	}
+
+	err = MountZFSDataset(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "not able to mount the dataset")
+	}
+
+	logrus.Infof("dataset %v mounted %v", volume, mount.MountPath)
+
+	return nil
+}
+
+// MountVolume mounts the disk to the specified path
+func MountVolume(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	switch vol.Spec.VolumeType {
+	case VOLTYPE_DATASET:
+		return MountDataset(vol, mount)
+	default:
+		return MountZvol(vol, mount)
+	}
+}
--- a/pkg/zfs/zfs_util.go
+++ b/pkg/zfs/zfs_util.go
@ -17,188 +17,337 @@ limitations under the License.
 package zfs

 import (
-	"os"
 	"os/exec"
+	"path/filepath"

 	"github.com/Sirupsen/logrus"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
 )

+// zfs related constants
+const (
+	ZFS_DEVPATH = "/dev/zvol/"
+	FSTYPE_ZFS  = "zfs"
+)
+
+// zfs command related constants
 const (
-	ZFS_DEVPATH   = "/dev/zvol/"
 	ZFSVolCmd     = "zfs"
 	ZFSCreateArg  = "create"
 	ZFSDestroyArg = "destroy"
 	ZFSSetArg     = "set"
+	ZFSListArg    = "list"
+)
+
+// constants to define volume type
+const (
+	VOLTYPE_DATASET = "DATASET"
+	VOLTYPE_ZVOL    = "ZVOL"
 )

 func PropertyChanged(oldVol *apis.ZFSVolume, newVol *apis.ZFSVolume) bool {
+	if oldVol.Spec.VolumeType == VOLTYPE_DATASET &&
+		newVol.Spec.VolumeType == VOLTYPE_DATASET &&
+		oldVol.Spec.RecordSize != newVol.Spec.RecordSize {
+		return true
+	}
+
 	return oldVol.Spec.Compression != newVol.Spec.Compression ||
-		oldVol.Spec.Dedup != newVol.Spec.Dedup ||
-		oldVol.Spec.Capacity != newVol.Spec.Capacity
+		oldVol.Spec.Dedup != newVol.Spec.Dedup
 }

-// builldVolumeCreateArgs returns zvol create command along with attributes as a string array
-func buildVolumeCreateArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+// GetVolumeType returns the volume type
+// whether it is a zvol or dataset
+func GetVolumeType(fstype string) string {
+	/*
+	 * if fstype is provided as zfs then a zfs dataset will be created
+	 * otherwise a zvol will be created
+	 */
+	switch fstype {
+	case FSTYPE_ZFS:
+		return VOLTYPE_DATASET
+	default:
+		return VOLTYPE_ZVOL
+	}
+}

-	zvol := vol.Spec.PoolName + "/" + vol.Name
+// builldZvolCreateArgs returns zfs create command for zvol along with attributes as a string array
+func buildZvolCreateArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string

-	ZFSVolCmd = append(ZFSVolCmd, ZFSCreateArg)
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	ZFSVolArg = append(ZFSVolArg, ZFSCreateArg)

 	if vol.Spec.ThinProvision == "yes" {
-		ZFSVolCmd = append(ZFSVolCmd, "-s")
+		ZFSVolArg = append(ZFSVolArg, "-s")
 	}
 	if len(vol.Spec.Capacity) != 0 {
-		ZFSVolCmd = append(ZFSVolCmd, "-V", vol.Spec.Capacity)
+		ZFSVolArg = append(ZFSVolArg, "-V", vol.Spec.Capacity)
 	}
-	if len(vol.Spec.BlockSize) != 0 {
-		ZFSVolCmd = append(ZFSVolCmd, "-b", vol.Spec.BlockSize)
+	if len(vol.Spec.VolBlockSize) != 0 {
+		ZFSVolArg = append(ZFSVolArg, "-b", vol.Spec.VolBlockSize)
 	}
 	if len(vol.Spec.Dedup) != 0 {
 		dedupProperty := "dedup=" + vol.Spec.Dedup
-		ZFSVolCmd = append(ZFSVolCmd, "-o", dedupProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", dedupProperty)
 	}
 	if len(vol.Spec.Compression) != 0 {
 		compressionProperty := "compression=" + vol.Spec.Compression
-		ZFSVolCmd = append(ZFSVolCmd, "-o", compressionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", compressionProperty)
 	}
 	if len(vol.Spec.Encryption) != 0 {
 		encryptionProperty := "encryption=" + vol.Spec.Encryption
-		ZFSVolCmd = append(ZFSVolCmd, "-o", encryptionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", encryptionProperty)
 	}
 	if len(vol.Spec.KeyLocation) != 0 {
 		keyLocation := "keylocation=" + vol.Spec.KeyLocation
-		ZFSVolCmd = append(ZFSVolCmd, "-o", keyLocation)
+		ZFSVolArg = append(ZFSVolArg, "-o", keyLocation)
 	}
 	if len(vol.Spec.KeyFormat) != 0 {
 		keyFormat := "keyformat=" + vol.Spec.KeyFormat
-		ZFSVolCmd = append(ZFSVolCmd, "-o", keyFormat)
+		ZFSVolArg = append(ZFSVolArg, "-o", keyFormat)
 	}

-	ZFSVolCmd = append(ZFSVolCmd, zvol)
+	ZFSVolArg = append(ZFSVolArg, volume)

-	return ZFSVolCmd
+	return ZFSVolArg
 }

-// builldVolumeSetArgs returns zvol set command along with attributes as a string array
-// TODO(pawan) need to find a way to identify which property has changed
-func buildVolumeSetArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+// builldDatasetCreateArgs returns zfs create command for dataset along with attributes as a string array
+func buildDatasetCreateArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string

-	zvol := vol.Spec.PoolName + "/" + vol.Name
+	volume := vol.Spec.PoolName + "/" + vol.Name

-	ZFSVolCmd = append(ZFSVolCmd, ZFSSetArg)
+	ZFSVolArg = append(ZFSVolArg, ZFSCreateArg)

 	if len(vol.Spec.Capacity) != 0 {
-		volsize := "volsize=" + vol.Spec.Capacity
-		ZFSVolCmd = append(ZFSVolCmd, volsize)
+		quotaProperty := "quota=" + vol.Spec.Capacity
+		ZFSVolArg = append(ZFSVolArg, "-o", quotaProperty)
+	}
+	if len(vol.Spec.RecordSize) != 0 {
+		recordsizeProperty := "recordsize=" + vol.Spec.RecordSize
+		ZFSVolArg = append(ZFSVolArg, "-o", recordsizeProperty)
+	}
+	if vol.Spec.ThinProvision == "no" {
+		reservationProperty := "reservation=" + vol.Spec.Capacity
+		ZFSVolArg = append(ZFSVolArg, "-o", reservationProperty)
 	}
 	if len(vol.Spec.Dedup) != 0 {
 		dedupProperty := "dedup=" + vol.Spec.Dedup
-		ZFSVolCmd = append(ZFSVolCmd, dedupProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", dedupProperty)
 	}
 	if len(vol.Spec.Compression) != 0 {
 		compressionProperty := "compression=" + vol.Spec.Compression
-		ZFSVolCmd = append(ZFSVolCmd, compressionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", compressionProperty)
+	}
+	if len(vol.Spec.Encryption) != 0 {
+		encryptionProperty := "encryption=" + vol.Spec.Encryption
+		ZFSVolArg = append(ZFSVolArg, "-o", encryptionProperty)
+	}
+	if len(vol.Spec.KeyLocation) != 0 {
+		keyLocation := "keylocation=" + vol.Spec.KeyLocation
+		ZFSVolArg = append(ZFSVolArg, "-o", keyLocation)
+	}
+	if len(vol.Spec.KeyFormat) != 0 {
+		keyFormat := "keyformat=" + vol.Spec.KeyFormat
+		ZFSVolArg = append(ZFSVolArg, "-o", keyFormat)
 	}

-	ZFSVolCmd = append(ZFSVolCmd, zvol)
+	// set the mount path to none, by default zfs mounts it to the default dataset path
+	ZFSVolArg = append(ZFSVolArg, "-o", "mountpoint=none", volume)

-	return ZFSVolCmd
+	return ZFSVolArg
 }

-// builldVolumeDestroyArgs returns zvol destroy command along with attributes as a string array
+// builldVolumeSetArgs returns volume set command along with attributes as a string array
+// TODO(pawan) need to find a way to identify which property has changed
+func buildVolumeSetArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string
+
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	ZFSVolArg = append(ZFSVolArg, ZFSSetArg)
+
+	if vol.Spec.VolumeType == VOLTYPE_DATASET &&
+		len(vol.Spec.RecordSize) != 0 {
+		recordsizeProperty := "recordsize=" + vol.Spec.RecordSize
+		ZFSVolArg = append(ZFSVolArg, recordsizeProperty)
+	}
+
+	if len(vol.Spec.Dedup) != 0 {
+		dedupProperty := "dedup=" + vol.Spec.Dedup
+		ZFSVolArg = append(ZFSVolArg, dedupProperty)
+	}
+	if len(vol.Spec.Compression) != 0 {
+		compressionProperty := "compression=" + vol.Spec.Compression
+		ZFSVolArg = append(ZFSVolArg, compressionProperty)
+	}
+
+	ZFSVolArg = append(ZFSVolArg, volume)
+
+	return ZFSVolArg
+}
+
+// builldVolumeDestroyArgs returns volume destroy command along with attributes as a string array
 func buildVolumeDestroyArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+	var ZFSVolArg []string

-	zvol := vol.Spec.PoolName + "/" + vol.Name
+	volume := vol.Spec.PoolName + "/" + vol.Name

-	ZFSVolCmd = append(ZFSVolCmd, ZFSDestroyArg, "-R", zvol)
+	ZFSVolArg = append(ZFSVolArg, ZFSDestroyArg, "-R", volume)

-	return ZFSVolCmd
+	return ZFSVolArg
 }

-// CreateZvol creates the zvol and returns the corresponding diskPath
-// of the volume which gets created on the node
-func CreateZvol(vol *apis.ZFSVolume) error {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+func getVolume(volume string) error {
+	var ZFSVolArg []string

-	if _, err := os.Stat(devicePath); os.IsNotExist(err) {
+	ZFSVolArg = append(ZFSVolArg, ZFSListArg, volume)

-		args := buildVolumeCreateArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, ZFSVolArg...)
+	_, err := cmd.CombinedOutput()
+	return err
+}
+
+// CreateVolume creates the zvol/dataset as per
+// info provided in ZFSVolume object
+func CreateVolume(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	if err := getVolume(volume); err != nil {
+		var args []string
+		if vol.Spec.VolumeType == VOLTYPE_DATASET {
+			args = buildDatasetCreateArgs(vol)
+		} else {
+			args = buildZvolCreateArgs(vol)
+		}
 		cmd := exec.Command(ZFSVolCmd, args...)
 		out, err := cmd.CombinedOutput()

 		if err != nil {
 			logrus.Errorf(
-				"zfs: could not create zvol %v cmd %v error: %s", zvol, args, string(out),
+				"zfs: could not create volume %v cmd %v error: %s", volume, args, string(out),
 			)
 			return err
 		}
-		logrus.Infof("created zvol %s", zvol)
+		logrus.Infof("created volume %s", volume)
 	} else if err == nil {
-		logrus.Infof("using existing zvol %v", zvol)
-	} else {
-		return err
+		logrus.Infof("using existing volume %v", volume)
 	}

 	return nil
 }

-// SetZvolProp sets the zvol property
+// SetDatasetMountProp sets mountpoint for the volume
+func SetDatasetMountProp(volume string, mountpath string) error {
+	var ZFSVolArg []string
+
+	mountProperty := "mountpoint=" + mountpath
+	ZFSVolArg = append(ZFSVolArg, ZFSSetArg, mountProperty, volume)
+
+	cmd := exec.Command(ZFSVolCmd, ZFSVolArg...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		logrus.Errorf("zfs: could not set mountpoint on dataset %v cmd %v error: %s",
+			volume, ZFSVolArg, string(out))
+	}
+	return err
+}
+
+// MountZFSDataset mounts the dataset to the given mountpoint
+func MountZFSDataset(vol *apis.ZFSVolume, mountpath string) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	return SetDatasetMountProp(volume, mountpath)
+}
+
+// UmountZFSDataset umounts the dataset
+func UmountZFSDataset(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	return SetDatasetMountProp(volume, "none")
+}
+
+// SetZvolProp sets the volume property
 func SetZvolProp(vol *apis.ZFSVolume) error {
 	var err error
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+	volume := vol.Spec.PoolName + "/" + vol.Name

-	if _, err = os.Stat(devicePath); err == nil {
-		args := buildVolumeSetArgs(vol)
-		cmd := exec.Command(ZFSVolCmd, args...)
-		out, err := cmd.CombinedOutput()
-
-		if err != nil {
-			logrus.Errorf(
-				"zfs: could not set property on zvol %v cmd %v error: %s", zvol, args, string(out),
-			)
-			return err
-		}
-		logrus.Infof("property set on zvol %s", zvol)
+	if len(vol.Spec.Compression) == 0 &&
+		len(vol.Spec.Dedup) == 0 &&
+		(vol.Spec.VolumeType != VOLTYPE_DATASET ||
+			len(vol.Spec.RecordSize) == 0) {
+		//nothing to set, just return
+		return nil
 	}
+	/* Case: Restart =>
+	 * In this case we get the add event but here we don't know which
+	 * property has changed when we were down, so firing the zfs set
+	 * command with the all property present on the ZFSVolume.
+
+	 * Case: Property Change =>
+	 * TODO(pawan) When we get the update event, we make sure at least
+	 * one property has changed before adding it to the event queue for
+	 * handling. At this stage, since we haven't stored the
+	 * ZFSVolume object as it will be too heavy, we are firing the set
+	 * command with the all property preset in the ZFSVolume object since
+	 * it is guaranteed that at least one property has changed.
+	 */
+
+	args := buildVolumeSetArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, args...)
+	out, err := cmd.CombinedOutput()
+
+	if err != nil {
+		logrus.Errorf(
+			"zfs: could not set property on volume %v cmd %v error: %s", volume, args, string(out),
+		)
+		return err
+	}
+	logrus.Infof("property set on volume %s", volume)

 	return err
 }

-// DestroyZvol deletes the zvol
-func DestroyZvol(vol *apis.ZFSVolume) error {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+// DestroyVolume deletes the zfs volume
+func DestroyVolume(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name

-	if _, err := os.Stat(devicePath); err == nil {
-		args := buildVolumeDestroyArgs(vol)
-		cmd := exec.Command(ZFSVolCmd, args...)
-		out, err := cmd.CombinedOutput()
-
-		if err != nil {
-			logrus.Errorf(
-				"zfs: could not destroy zvol %v cmd %v error: %s", zvol, args, string(out),
-			)
-			return err
-		}
-		logrus.Infof("destroyed zvol %s", zvol)
+	if err := getVolume(volume); err != nil {
+		return nil
 	}

+	args := buildVolumeDestroyArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, args...)
+	out, err := cmd.CombinedOutput()
+
+	if err != nil {
+		logrus.Errorf(
+			"zfs: could not destroy volume %v cmd %v error: %s", volume, args, string(out),
+		)
+		return err
+	}
+	logrus.Infof("destroyed volume %s", volume)
+
 	return nil
 }

-// GetDevicePath returns device path for zvol if it exists
-func GetDevicePath(vol *apis.ZFSVolume) (string, error) {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+// GetVolumeDevPath returns devpath for the given volume
+func GetVolumeDevPath(vol *apis.ZFSVolume) (string, error) {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	if vol.Spec.VolumeType == VOLTYPE_DATASET {
+		return volume, nil
+	}

-	if _, err := os.Stat(devicePath); os.IsNotExist(err) {
+	devicePath := ZFS_DEVPATH + volume
+
+	// evaluate the symlink to get the dev path for zvol
+	dev, err := filepath.EvalSymlinks(devicePath)
+	if err != nil {
 		return "", err
 	}
-	return devicePath, nil
+
+	return dev, nil
 }