feat(resize): adding Online volume expansion support for ZFSPV

We can resize the volume by updating the PVC yaml to
the desired size and apply it. The ZFS Driver will take care
of updating the quota in case of dataset. If we are using a
Zvol and have mounted it as ext4 or xfs filesystem, the driver will take
care of expanding the volume via reize2fs/xfs_growfs binaries.

For resize, storageclass that provisions the pvc must suppo
rt resize. We should have allowVolumeExpansion as true in storageclass

```yaml
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: openebs-zfspv
allowVolumeExpansion: true
parameters:
  poolname: "zfspv-pool"
provisioner: zfs.csi.openebs.io

```

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

pkg/driver/agent.go

@@ -220,6 +220,13 @@ func (ns *node) NodeGetCapabilities(
 					},
 				},
 			},
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+					},
+				},
+			},
 		},
 	}, nil
 }
@@ -267,7 +274,29 @@ func (ns *node) NodeExpandVolume(
 	req *csi.NodeExpandVolumeRequest,
 ) (*csi.NodeExpandVolumeResponse, error) {
 
-	return nil, status.Error(codes.Unimplemented, "")
+	volumeID := req.GetVolumeId()
+	vol, err := zfs.GetZFSVolume(volumeID)
+
+	if err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"failed to handle NodeExpandVolume Request for %s, {%s}",
+			req.VolumeId,
+			err.Error(),
+		)
+	}
+	if err = zfs.ResizeZFSVolume(vol, req.GetVolumePath()); err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"failed to handle NodeExpandVolume Request for %s, {%s}",
+			req.VolumeId,
+			err.Error(),
+		)
+	}
+
+	return &csi.NodeExpandVolumeResponse{
+		CapacityBytes: req.GetCapacityRange().GetRequiredBytes(),
+	}, nil
 }
 
 // NodeGetVolumeStats returns statistics for the

pkg/driver/controller.go

@@ -291,7 +291,50 @@ func (cs *controller) ControllerExpandVolume(
 	req *csi.ControllerExpandVolumeRequest,
 ) (*csi.ControllerExpandVolumeResponse, error) {
 
-	return nil, status.Error(codes.Unimplemented, "")
+	updatedSize := req.GetCapacityRange().GetRequiredBytes()
+
+	vol, err := zfs.GetZFSVolume(req.VolumeId)
+	if err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"ControllerExpandVolumeRequest: failed to get ZFSVolume in for %s, {%s}",
+			req.VolumeId,
+			err.Error(),
+		)
+	}
+
+	volsize, err := strconv.ParseInt(vol.Spec.Capacity, 10, 64)
+	if err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"ControllerExpandVolumeRequest: failed to parse volsize in for %s, {%s}",
+			req.VolumeId,
+			err.Error(),
+		)
+	}
+	/*
+	 * Controller expand volume must be idempotent. If a volume corresponding
+	 * to the specified volume ID is already larger than or equal to the target
+	 * capacity of the expansion request, the plugin should reply 0 OK.
+	 */
+	if volsize >= updatedSize {
+		return csipayload.NewControllerExpandVolumeResponseBuilder().
+			WithCapacityBytes(volsize).
+			Build(), nil
+	}
+
+	if err := zfs.ResizeVolume(vol, updatedSize); err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"failed to handle ControllerExpandVolumeRequest for %s, {%s}",
+			req.VolumeId,
+			err.Error(),
+		)
+	}
+	return csipayload.NewControllerExpandVolumeResponseBuilder().
+		WithCapacityBytes(updatedSize).
+		WithNodeExpansionRequired(true).
+		Build(), nil
 }
 
 // CreateSnapshot creates a snapshot for given volume
@@ -521,6 +564,7 @@ func newControllerCapabilities() []*csi.ControllerServiceCapability {
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 		csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
+		csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
 	} {
 		capabilities = append(capabilities, fromType(cap))
 	}