feat(zfs-localpv): initial commit

provisioning and deprovisioning of
the volumes on the node where zfs pool
has already been setup. Pool name and the volume
parameters has to be given in storage class
which will be used to provision the volume.

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

pkg/driver/agent.go

@@ -0,0 +1,297 @@
+/*
+Copyright © 2019 The OpenEBS Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"github.com/Sirupsen/logrus"
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	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"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sync"
+)
+
+// node is the server implementation
+// for CSI NodeServer
+type node struct {
+	driver *CSIDriver
+}
+
+// NewNode returns a new instance
+// of CSI NodeServer
+func NewNode(d *CSIDriver) csi.NodeServer {
+	var ControllerMutex = sync.RWMutex{}
+	// start the zfsvolume watcher
+	go func() {
+		err := ctrl.Start(&ControllerMutex)
+		if err != nil {
+			logrus.Errorf("Failed to start cstorvolume claim controller: %s", err.Error())
+		}
+	}()
+
+	return &node{
+		driver: d,
+	}
+}
+
+func GetVolAndMountInfo(
+	req *csi.NodePublishVolumeRequest,
+) (*apis.ZFSVolume, *apis.MountInfo, error) {
+	var mountinfo apis.MountInfo
+
+	mountinfo.FSType = req.GetVolumeCapability().GetMount().GetFsType()
+	mountinfo.MountPath = req.GetTargetPath()
+	mountinfo.ReadOnly = req.GetReadonly()
+	mountinfo.MountOptions = append(mountinfo.MountOptions, req.GetVolumeCapability().GetMount().GetMountFlags()...)
+
+	getOptions := metav1.GetOptions{}
+	vol, err := builder.NewKubeclient().
+		WithNamespace(zvol.OpenEBSNamespace).
+		Get(req.GetVolumeId(), getOptions)
+
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return vol, &mountinfo, nil
+}
+
+// NodePublishVolume publishes (mounts) the volume
+// at the corresponding node at a given path
+//
+// This implements csi.NodeServer
+func (ns *node) NodePublishVolume(
+	ctx context.Context,
+	req *csi.NodePublishVolumeRequest,
+) (*csi.NodePublishVolumeResponse, error) {
+
+	var (
+		err error
+	)
+
+	if err = ns.validateNodePublishReq(req); err != nil {
+		return nil, err
+	}
+
+	vol, mountInfo, err := GetVolAndMountInfo(req)
+	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 {
+		goto PublishVolumeResponse
+	}
+
+PublishVolumeResponse:
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	return &csi.NodePublishVolumeResponse{}, nil
+}
+
+// NodeUnpublishVolume unpublishes (unmounts) the volume
+// from the corresponding node from the given path
+//
+// This implements csi.NodeServer
+func (ns *node) NodeUnpublishVolume(
+	ctx context.Context,
+	req *csi.NodeUnpublishVolumeRequest,
+) (*csi.NodeUnpublishVolumeResponse, error) {
+
+	var (
+		err           error
+		vol           *apis.ZFSVolume
+		currentMounts []string
+	)
+
+	if err = ns.validateNodeUnpublishReq(req); err != nil {
+		return nil, err
+	}
+
+	targetPath := req.GetTargetPath()
+	volumeID := req.GetVolumeId()
+
+	getOptions := metav1.GetOptions{}
+	vol, err = builder.NewKubeclient().
+		WithNamespace(zvol.OpenEBSNamespace).
+		Get(volumeID, getOptions)
+
+	if err != nil {
+		return nil, err
+	}
+
+	zfsvolume := vol.Spec.PoolName + "/" + vol.Name
+	devpath := zvol.ZFS_DEVPATH + zfsvolume
+	currentMounts, err = zvol.GetMounts(devpath)
+	if err != nil {
+		return nil, err
+	} else if len(currentMounts) == 0 {
+		goto NodeUnpublishResponse
+	} else if len(currentMounts) == 1 {
+		if currentMounts[0] != targetPath {
+			return nil, status.Error(codes.Internal, "device not mounted at right path")
+		}
+	} else {
+		logrus.Errorf(
+			"can not unmount, more than one mounts for volume:%s path %s mounts: %v",
+			volumeID, targetPath, currentMounts,
+		)
+		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 {
+		goto NodeUnpublishResponse
+	}
+
+NodeUnpublishResponse:
+	logrus.Infof("hostpath: volume %s path: %s has been unmounted.",
+		volumeID, targetPath)
+
+	return &csi.NodeUnpublishVolumeResponse{}, nil
+}
+
+// NodeGetInfo returns node details
+//
+// This implements csi.NodeServer
+func (ns *node) NodeGetInfo(
+	ctx context.Context,
+	req *csi.NodeGetInfoRequest,
+) (*csi.NodeGetInfoResponse, error) {
+
+	return &csi.NodeGetInfoResponse{
+		NodeId: ns.driver.config.NodeID,
+	}, nil
+}
+
+// NodeGetCapabilities returns capabilities supported
+// by this node service
+//
+// This implements csi.NodeServer
+func (ns *node) NodeGetCapabilities(
+	ctx context.Context,
+	req *csi.NodeGetCapabilitiesRequest,
+) (*csi.NodeGetCapabilitiesResponse, error) {
+
+	return &csi.NodeGetCapabilitiesResponse{
+		Capabilities: []*csi.NodeServiceCapability{
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_UNKNOWN,
+					},
+				},
+			},
+		},
+	}, nil
+}
+
+// TODO
+// This needs to be implemented
+//
+// NodeStageVolume mounts the volume on the staging
+// path
+//
+// This implements csi.NodeServer
+func (ns *node) NodeStageVolume(
+	ctx context.Context,
+	req *csi.NodeStageVolumeRequest,
+) (*csi.NodeStageVolumeResponse, error) {
+
+	return &csi.NodeStageVolumeResponse{}, nil
+}
+
+// NodeUnstageVolume unmounts the volume from
+// the staging path
+//
+// This implements csi.NodeServer
+func (ns *node) NodeUnstageVolume(
+	ctx context.Context,
+	req *csi.NodeUnstageVolumeRequest,
+) (*csi.NodeUnstageVolumeResponse, error) {
+
+	return &csi.NodeUnstageVolumeResponse{}, nil
+}
+
+// TODO
+// Verify if this needs to be implemented
+//
+// NodeExpandVolume resizes the filesystem if required
+//
+// If ControllerExpandVolumeResponse returns true in
+// node_expansion_required then FileSystemResizePending
+// condition will be added to PVC and NodeExpandVolume
+// operation will be queued on kubelet
+//
+// This implements csi.NodeServer
+func (ns *node) NodeExpandVolume(
+	ctx context.Context,
+	req *csi.NodeExpandVolumeRequest,
+) (*csi.NodeExpandVolumeResponse, error) {
+
+	return nil, nil
+}
+
+// NodeGetVolumeStats returns statistics for the
+// given volume
+//
+// This implements csi.NodeServer
+func (ns *node) NodeGetVolumeStats(
+	ctx context.Context,
+	in *csi.NodeGetVolumeStatsRequest,
+) (*csi.NodeGetVolumeStatsResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (ns *node) validateNodePublishReq(
+	req *csi.NodePublishVolumeRequest,
+) error {
+	if req.GetVolumeCapability() == nil {
+		return status.Error(codes.InvalidArgument,
+			"Volume capability missing in request")
+	}
+
+	if len(req.GetVolumeId()) == 0 {
+		return status.Error(codes.InvalidArgument,
+			"Volume ID missing in request")
+	}
+	return nil
+}
+
+func (ns *node) validateNodeUnpublishReq(
+	req *csi.NodeUnpublishVolumeRequest,
+) error {
+	if req.GetVolumeId() == "" {
+		return status.Error(codes.InvalidArgument,
+			"Volume ID missing in request")
+	}
+
+	if req.GetTargetPath() == "" {
+		return status.Error(codes.InvalidArgument,
+			"Target path missing in request")
+	}
+	return nil
+}

pkg/driver/controller.go

@@ -0,0 +1,375 @@
+/*
+Copyright © 2019 The OpenEBS Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"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"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// controller is the server implementation
+// for CSI Controller
+type controller struct {
+	driver       *CSIDriver
+	capabilities []*csi.ControllerServiceCapability
+}
+
+// NewController returns a new instance
+// of CSI controller
+func NewController(d *CSIDriver) csi.ControllerServer {
+	return &controller{
+		driver:       d,
+		capabilities: newControllerCapabilities(),
+	}
+}
+
+// SupportedVolumeCapabilityAccessModes contains the list of supported access
+// modes for the volume
+var SupportedVolumeCapabilityAccessModes = []*csi.VolumeCapability_AccessMode{
+	&csi.VolumeCapability_AccessMode{
+		Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+	},
+}
+
+// CreateVolume provisions a volume
+func (cs *controller) CreateVolume(
+	ctx context.Context,
+	req *csi.CreateVolumeRequest,
+) (*csi.CreateVolumeResponse, error) {
+
+	logrus.Infof("received request to create volume {%s} vol{%v}", req.GetName(), req)
+	var err error
+
+	if err = cs.validateVolumeCreateReq(req); err != nil {
+		return nil, err
+	}
+
+	volName := req.GetName()
+	size := req.GetCapacityRange().RequiredBytes
+	bs := req.GetParameters()["blocksize"]
+	compression := req.GetParameters()["compression"]
+	dedup := req.GetParameters()["dedup"]
+	pool := req.GetParameters()["poolname"]
+	tp := req.GetParameters()["thinprovision"]
+
+	volObj, err := builder.NewBuilder().
+		WithName(volName).
+		WithCapacity(strconv.FormatInt(int64(size), 10)).
+		WithBlockSize(bs).
+		WithPoolName(pool).
+		WithDedup(dedup).
+		WithThinProv(tp).
+		WithCompression(compression).Build()
+
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	err = zvol.ProvisionVolume(size, volObj)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	return csipayload.NewCreateVolumeResponseBuilder().
+		WithName(volName).
+		WithCapacity(size).
+		Build(), nil
+}
+
+// DeleteVolume deletes the specified volume
+func (cs *controller) DeleteVolume(
+	ctx context.Context,
+	req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+
+	logrus.Infof("received request to delete volume {%s}", req.VolumeId)
+
+	var (
+		err error
+	)
+
+	if err = cs.validateDeleteVolumeReq(req); err != nil {
+		return nil, err
+	}
+
+	volumeID := req.GetVolumeId()
+
+	// verify if the volume has already been deleted
+	vol, err := zvol.GetVolume(volumeID)
+	if vol != nil && vol.DeletionTimestamp != nil {
+		goto deleteResponse
+	}
+
+	// Delete the corresponding ZV CR
+	err = zvol.DeleteVolume(volumeID)
+	if err != nil {
+		return nil, errors.Wrapf(
+			err,
+			"failed to handle delete volume request for {%s}",
+			volumeID,
+		)
+	}
+deleteResponse:
+	return csipayload.NewDeleteVolumeResponseBuilder().Build(), nil
+}
+
+// TODO Implementation will be taken up later
+
+// ValidateVolumeCapabilities validates the capabilities
+// required to create a new volume
+// This implements csi.ControllerServer
+func (cs *controller) ValidateVolumeCapabilities(
+	ctx context.Context,
+	req *csi.ValidateVolumeCapabilitiesRequest,
+) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// ControllerGetCapabilities fetches controller capabilities
+//
+// This implements csi.ControllerServer
+func (cs *controller) ControllerGetCapabilities(
+	ctx context.Context,
+	req *csi.ControllerGetCapabilitiesRequest,
+) (*csi.ControllerGetCapabilitiesResponse, error) {
+
+	resp := &csi.ControllerGetCapabilitiesResponse{
+		Capabilities: cs.capabilities,
+	}
+
+	return resp, nil
+}
+
+// ControllerExpandVolume resizes previously provisioned volume
+//
+// This implements csi.ControllerServer
+func (cs *controller) ControllerExpandVolume(
+	ctx context.Context,
+	req *csi.ControllerExpandVolumeRequest,
+) (*csi.ControllerExpandVolumeResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// CreateSnapshot creates a snapshot for given volume
+//
+// This implements csi.ControllerServer
+func (cs *controller) CreateSnapshot(
+	ctx context.Context,
+	req *csi.CreateSnapshotRequest,
+) (*csi.CreateSnapshotResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// DeleteSnapshot deletes given snapshot
+//
+// This implements csi.ControllerServer
+func (cs *controller) DeleteSnapshot(
+	ctx context.Context,
+	req *csi.DeleteSnapshotRequest,
+) (*csi.DeleteSnapshotResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// ListSnapshots lists all snapshots for the
+// given volume
+//
+// This implements csi.ControllerServer
+func (cs *controller) ListSnapshots(
+	ctx context.Context,
+	req *csi.ListSnapshotsRequest,
+) (*csi.ListSnapshotsResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// ControllerUnpublishVolume removes a previously
+// attached volume from the given node
+//
+// This implements csi.ControllerServer
+func (cs *controller) ControllerUnpublishVolume(
+	ctx context.Context,
+	req *csi.ControllerUnpublishVolumeRequest,
+) (*csi.ControllerUnpublishVolumeResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// ControllerPublishVolume attaches given volume
+// at the specified node
+//
+// This implements csi.ControllerServer
+func (cs *controller) ControllerPublishVolume(
+	ctx context.Context,
+	req *csi.ControllerPublishVolumeRequest,
+) (*csi.ControllerPublishVolumeResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// GetCapacity return the capacity of the
+// given volume
+//
+// This implements csi.ControllerServer
+func (cs *controller) GetCapacity(
+	ctx context.Context,
+	req *csi.GetCapacityRequest,
+) (*csi.GetCapacityResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// ListVolumes lists all the volumes
+//
+// This implements csi.ControllerServer
+func (cs *controller) ListVolumes(
+	ctx context.Context,
+	req *csi.ListVolumesRequest,
+) (*csi.ListVolumesResponse, error) {
+
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+// validateCapabilities validates if provided capabilities
+// are supported by this driver
+func validateCapabilities(caps []*csi.VolumeCapability) bool {
+
+	for _, cap := range caps {
+		if !IsSupportedVolumeCapabilityAccessMode(cap.AccessMode.Mode) {
+			return false
+		}
+	}
+	return true
+}
+
+func (cs *controller) validateDeleteVolumeReq(req *csi.DeleteVolumeRequest) error {
+	volumeID := req.GetVolumeId()
+	if volumeID == "" {
+		return status.Error(
+			codes.InvalidArgument,
+			"failed to handle delete volume request: missing volume id",
+		)
+	}
+
+	err := cs.validateRequest(
+		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+	)
+	if err != nil {
+		return errors.Wrapf(
+			err,
+			"failed to handle delete volume request for {%s} : validation failed",
+			volumeID,
+		)
+	}
+	return nil
+}
+
+// IsSupportedVolumeCapabilityAccessMode valides the requested access mode
+func IsSupportedVolumeCapabilityAccessMode(
+	accessMode csi.VolumeCapability_AccessMode_Mode,
+) bool {
+
+	for _, access := range SupportedVolumeCapabilityAccessModes {
+		if accessMode == access.Mode {
+			return true
+		}
+	}
+	return false
+}
+
+// newControllerCapabilities returns a list
+// of this controller's capabilities
+func newControllerCapabilities() []*csi.ControllerServiceCapability {
+	fromType := func(
+		cap csi.ControllerServiceCapability_RPC_Type,
+	) *csi.ControllerServiceCapability {
+		return &csi.ControllerServiceCapability{
+			Type: &csi.ControllerServiceCapability_Rpc{
+				Rpc: &csi.ControllerServiceCapability_RPC{
+					Type: cap,
+				},
+			},
+		}
+	}
+
+	var capabilities []*csi.ControllerServiceCapability
+	for _, cap := range []csi.ControllerServiceCapability_RPC_Type{
+		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+	} {
+		capabilities = append(capabilities, fromType(cap))
+	}
+	return capabilities
+}
+
+// validateRequest validates if the requested service is
+// supported by the driver
+func (cs *controller) validateRequest(
+	c csi.ControllerServiceCapability_RPC_Type,
+) error {
+
+	for _, cap := range cs.capabilities {
+		if c == cap.GetRpc().GetType() {
+			return nil
+		}
+	}
+
+	return status.Error(
+		codes.InvalidArgument,
+		fmt.Sprintf("failed to validate request: {%s} is not supported", c),
+	)
+}
+
+func (cs *controller) validateVolumeCreateReq(req *csi.CreateVolumeRequest) error {
+	err := cs.validateRequest(
+		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+	)
+	if err != nil {
+		return errors.Wrapf(
+			err,
+			"failed to handle create volume request for {%s}",
+			req.GetName(),
+		)
+	}
+
+	if req.GetName() == "" {
+		return status.Error(
+			codes.InvalidArgument,
+			"failed to handle create volume request: missing volume name",
+		)
+	}
+
+	volCapabilities := req.GetVolumeCapabilities()
+	if volCapabilities == nil {
+		return status.Error(
+			codes.InvalidArgument,
+			"failed to handle create volume request: missing volume capabilities",
+		)
+	}
+	return nil
+}

pkg/driver/driver.go

@@ -0,0 +1,104 @@
+/*
+Copyright © 2019 The OpenEBS Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"github.com/Sirupsen/logrus"
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	config "github.com/openebs/zfs-localpv/pkg/config"
+)
+
+// volume can only be published once as
+// read/write on a single node, at any
+// given time
+var supportedAccessMode = &csi.VolumeCapability_AccessMode{
+	Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+}
+
+// TODO check if this can be renamed to Base
+//
+// CSIDriver defines a common data structure
+// for drivers
+type CSIDriver struct {
+	// TODO change the field names to make it
+	// readable
+	config *config.Config
+	ids    csi.IdentityServer
+	ns     csi.NodeServer
+	cs     csi.ControllerServer
+
+	cap []*csi.VolumeCapability_AccessMode
+}
+
+// GetVolumeCapabilityAccessModes fetches the access
+// modes on which the volume can be exposed
+func GetVolumeCapabilityAccessModes() []*csi.VolumeCapability_AccessMode {
+	supported := []csi.VolumeCapability_AccessMode_Mode{
+		csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+	}
+
+	var vcams []*csi.VolumeCapability_AccessMode
+	for _, vcam := range supported {
+		logrus.Infof("enabling volume access mode: %s", vcam.String())
+		vcams = append(vcams, newVolumeCapabilityAccessMode(vcam))
+	}
+	return vcams
+}
+
+func newVolumeCapabilityAccessMode(mode csi.VolumeCapability_AccessMode_Mode) *csi.VolumeCapability_AccessMode {
+	return &csi.VolumeCapability_AccessMode{Mode: mode}
+}
+
+// New returns a new driver instance
+func New(config *config.Config) *CSIDriver {
+	driver := &CSIDriver{
+		config: config,
+		cap:    GetVolumeCapabilityAccessModes(),
+	}
+
+	switch config.PluginType {
+	case "controller":
+		driver.cs = NewController(driver)
+
+	case "agent":
+		// Start monitor goroutine to monitor the
+		// ZfsVolume CR. If there is any event
+		// related to the volume like destroy or
+		// property change, handle it accordingly.
+
+		driver.ns = NewNode(driver)
+	}
+
+	// Identity server is common to both node and
+	// controller, it is required to register,
+	// share capabilities and probe the corresponding
+	// driver
+	driver.ids = NewIdentity(driver)
+	return driver
+}
+
+// Run starts the CSI plugin by communicating
+// over the given endpoint
+func (d *CSIDriver) Run() error {
+	// Initialize and start listening on grpc server
+	s := NewNonBlockingGRPCServer(d.config.Endpoint, d.ids, d.cs, d.ns)
+
+	s.Start()
+	s.Wait()
+
+	return nil
+}

pkg/driver/grpc.go

@@ -0,0 +1,170 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"sync"
+
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+)
+
+// parseEndpoint should have a valid prefix(unix/tcp) to return a valid endpoint parts
+func parseEndpoint(ep string) (string, string, error) {
+	if strings.HasPrefix(strings.ToLower(ep), "unix://") || strings.HasPrefix(strings.ToLower(ep), "tcp://") {
+		s := strings.SplitN(ep, "://", 2)
+		if s[1] != "" {
+			return s[0], s[1], nil
+		}
+	}
+	return "", "", fmt.Errorf("Invalid endpoint: %v", ep)
+}
+
+// logGRPC logs all the grpc related errors, i.e the final errors
+// which are returned to the grpc clients
+func logGRPC(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	logrus.Infof("GRPC call: %s", info.FullMethod)
+	logrus.Infof("GRPC request: %s", protosanitizer.StripSecrets(req))
+	resp, err := handler(ctx, req)
+	if err != nil {
+		logrus.Errorf("GRPC error: %v", err)
+	} else {
+		logrus.Infof("GRPC response: %s", protosanitizer.StripSecrets(resp))
+	}
+	return resp, err
+}
+
+// NonBlockingGRPCServer defines Non blocking GRPC server interfaces
+type NonBlockingGRPCServer interface {
+	// Start services at the endpoint
+	Start()
+
+	// Waits for the service to stop
+	Wait()
+
+	// Stops the service gracefully
+	Stop()
+
+	// Stops the service forcefully
+	ForceStop()
+}
+
+// NewNonBlockingGRPCServer returns a new instance of NonBlockingGRPCServer
+func NewNonBlockingGRPCServer(ep string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) NonBlockingGRPCServer {
+	return &nonBlockingGRPCServer{
+		endpoint:     ep,
+		idnty_server: ids,
+		ctrl_server:  cs,
+		agent_server: ns}
+}
+
+// NonBlocking server
+// dont block the execution for a task to complete.
+// use wait group to wait for all the tasks dispatched.
+type nonBlockingGRPCServer struct {
+	wg           sync.WaitGroup
+	server       *grpc.Server
+	endpoint     string
+	idnty_server csi.IdentityServer
+	ctrl_server  csi.ControllerServer
+	agent_server csi.NodeServer
+}
+
+// Start grpc server for serving CSI endpoints
+func (s *nonBlockingGRPCServer) Start() {
+
+	s.wg.Add(1)
+
+	go s.serve(s.endpoint, s.idnty_server, s.ctrl_server, s.agent_server)
+
+	return
+}
+
+// Wait for the service to stop
+func (s *nonBlockingGRPCServer) Wait() {
+	s.wg.Wait()
+}
+
+// Stop the service forcefully
+func (s *nonBlockingGRPCServer) Stop() {
+	s.server.GracefulStop()
+}
+
+// ForceStop the service
+func (s *nonBlockingGRPCServer) ForceStop() {
+	s.server.Stop()
+}
+
+// serve starts serving requests at the provided endpoint based on the type of
+// plugin. In this function all the csi related interfaces are provided by
+// container-storage-interface
+func (s *nonBlockingGRPCServer) serve(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
+
+	proto, addr, err := parseEndpoint(endpoint)
+	if err != nil {
+		logrus.Fatal(err.Error())
+	}
+
+	// Clear off the addr if it is already present, this is done to remove stale
+	// entries, as this path is shared with the OS and will be the same
+	// everytime the plugin restarts, its possible that the last instance leaves
+	// a stale entry
+	if proto == "unix" {
+		addr = "/" + addr
+		if err := os.Remove(addr); err != nil && !os.IsNotExist(err) {
+			logrus.Fatalf("Failed to remove %s, error: %s", addr, err.Error())
+		}
+	}
+
+	listener, err := net.Listen(proto, addr)
+	if err != nil {
+		logrus.Fatalf("Failed to listen: %v", err)
+	}
+
+	opts := []grpc.ServerOption{
+		grpc.UnaryInterceptor(logGRPC),
+	}
+	// Create a new grpc server, all the request from csi client to
+	// create/delete/... will hit this server
+	server := grpc.NewServer(opts...)
+	s.server = server
+
+	if ids != nil {
+		csi.RegisterIdentityServer(server, ids)
+	}
+	if cs != nil {
+		csi.RegisterControllerServer(server, cs)
+	}
+	if ns != nil {
+		csi.RegisterNodeServer(server, ns)
+	}
+
+	logrus.Infof("Listening for connections on address: %#v", listener.Addr())
+
+	// Start serving requests on the grpc server created
+	server.Serve(listener)
+
+}

pkg/driver/identity.go

@@ -0,0 +1,112 @@
+/*
+Copyright © 2019 The OpenEBS Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/openebs/zfs-localpv/pkg/version"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// identity is the server implementation
+// for CSI IdentityServer
+type identity struct {
+	driver *CSIDriver
+}
+
+// NewIdentity returns a new instance of CSI
+// IdentityServer
+func NewIdentity(d *CSIDriver) csi.IdentityServer {
+	return &identity{
+		driver: d,
+	}
+}
+
+// GetPluginInfo returns the version and name of
+// this service
+//
+// This implements csi.IdentityServer
+func (id *identity) GetPluginInfo(
+	ctx context.Context,
+	req *csi.GetPluginInfoRequest,
+) (*csi.GetPluginInfoResponse, error) {
+
+	if id.driver.config.DriverName == "" {
+		return nil, status.Error(codes.Unavailable, "missing driver name")
+	}
+
+	if id.driver.config.Version == "" {
+		return nil, status.Error(codes.Unavailable, "missing driver version")
+	}
+
+	return &csi.GetPluginInfoResponse{
+		Name: id.driver.config.DriverName,
+		// TODO
+		// verify which version needs to be used:
+		// config.version or version.Current()
+		VendorVersion: version.Current(),
+	}, nil
+}
+
+// TODO
+// Need to implement this
+//
+// Probe checks if the plugin is running or not
+//
+// This implements csi.IdentityServer
+func (id *identity) Probe(
+	ctx context.Context,
+	req *csi.ProbeRequest,
+) (*csi.ProbeResponse, error) {
+
+	return &csi.ProbeResponse{}, nil
+}
+
+// GetPluginCapabilities returns supported capabilities
+// of this plugin
+//
+// Currently it reports whether this plugin can serve
+// the Controller interface. Controller interface methods
+// are called dependant on this
+//
+// This implements csi.IdentityServer
+func (id *identity) GetPluginCapabilities(
+	ctx context.Context,
+	req *csi.GetPluginCapabilitiesRequest,
+) (*csi.GetPluginCapabilitiesResponse, error) {
+
+	return &csi.GetPluginCapabilitiesResponse{
+		Capabilities: []*csi.PluginCapability{
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
+					},
+				},
+			},
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
+		},
+	}, nil
+}