adding topology support for zfspv (#7)

This PR adds support to allow the CSI driver to pick up a node matching the  topology specified in the storage class. Admin can specify allowedTopologies in the StorageClass to specify the nodes where the zfs pools are setup

```yaml
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: openebs-zfspv
allowVolumeExpansion: true
parameters:
  blocksize: "4k"
  compression: "on"
  dedup: "on"
  thinprovision: "yes"
  poolname: "zfspv-pool"
provisioner: zfs-localpv
volumeBindingMode: WaitForFirstConsumer
allowedTopologies:
- matchLabelExpressions:
  - key: kubernetes.io/hostname
    values:
      - gke-zfspv-pawan-default-pool-c8929518-cgd4
      - gke-zfspv-pawan-default-pool-c8929518-dxzc
```

Note: This PR picks up the first node from the list of nodes available.

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

pkg/zfs/mount.go

@@ -109,14 +109,9 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 		return status.Error(codes.Internal, "volume is owned by different node")
 	}
 
-	devicePath, err := createZvol(vol)
+	devicePath, err := GetDevicePath(vol)
 	if err != nil {
-		return status.Error(codes.Internal, err.Error())
-	}
-
-	err = UpdateZvolInfo(vol)
-	if err != nil {
-		return status.Error(codes.Internal, err.Error())
+		return status.Error(codes.Internal, "not able to get the device path")
 	}
 
 	/*
@@ -138,7 +133,7 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 	}
 	err = FormatAndMountZvol(devicePath, mount)
 	if err != nil {
-		return status.Error(codes.Internal, err.Error())
+		return status.Error(codes.Internal, "not able to mount the volume")
 	}
 
 	return err

pkg/zfs/volume.go

@@ -31,9 +31,10 @@ const (
 	OpenEBSNamespaceKey string = "OPENEBS_NAMESPACE"
 	// ZFSFinalizer for the ZfsVolume CR
 	ZFSFinalizer string = "zfs.openebs.io/finalizer"
-	// ZFSNodeKey will be used to insert Label
-	// in ZfsVolume CR
+	// ZFSNodeKey will be used to insert Label in ZfsVolume CR
 	ZFSNodeKey string = "kubernetes.io/nodename"
+	// ZFSTopologyKey is supported topology key for the zfs driver
+	ZFSTopologyKey string = "kubernetes.io/hostname"
 )
 
 var (

pkg/zfs/zfs_util.go

@@ -119,9 +119,9 @@ func buildVolumeDestroyArgs(vol *apis.ZFSVolume) []string {
 	return ZFSVolCmd
 }
 
-// createZvol creates the zvol and returns the corresponding diskPath
+// CreateZvol creates the zvol and returns the corresponding diskPath
 // of the volume which gets created on the node
-func createZvol(vol *apis.ZFSVolume) (string, error) {
+func CreateZvol(vol *apis.ZFSVolume) error {
 	zvol := vol.Spec.PoolName + "/" + vol.Name
 	devicePath := ZFS_DEVPATH + zvol
 
@@ -135,16 +135,16 @@ func createZvol(vol *apis.ZFSVolume) (string, error) {
 			logrus.Errorf(
 				"zfs: could not create zvol %v cmd %v error: %s", zvol, args, string(out),
 			)
-			return "", err
+			return err
 		}
 		logrus.Infof("created zvol %s", zvol)
 	} else if err == nil {
 		logrus.Infof("using existing zvol %v", zvol)
 	} else {
-		return "", err
+		return err
 	}
 
-	return devicePath, nil
+	return nil
 }
 
 // SetZvolProp sets the zvol property
@@ -191,3 +191,14 @@ func DestroyZvol(vol *apis.ZFSVolume) error {
 
 	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
+
+	if _, err := os.Stat(devicePath); os.IsNotExist(err) {
+		return "", err
+	}
+	return devicePath, nil
+}