func (s *service) createFilesystemFromSnapshot(ctx context.Context, snapID, volumeName, arrayID string) (*types.Snapshot, error) {
	ctx, _, rid := GetRunidLog(ctx)
	unity, err := s.getUnityClient(ctx, arrayID)
	if err != nil {
		return nil, err
	}
	snapshotAPI := gounity.NewSnapshot(unity)
	snapResp, err := snapshotAPI.CopySnapshot(ctx, snapID, volumeName)
	if err != nil {
		return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Create Filesystem from snapshot failed with error. Error: %v", err))
	}
	snapResp, err = snapshotAPI.FindSnapshotByName(ctx, volumeName)
	if err != nil {
		return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Create Filesystem from snapshot failed with error. Error: %v", err))
	}
	return snapResp, nil
}
Create Volume from Snapshot(Copy snapshot on array)
package service
import (
	"fmt"
	"strconv"
	"strings"
	"github.com/dell/gounity/api"
	"github.com/dell/gounity/util"
	"github.com/golang/protobuf/ptypes"
	"github.com/golang/protobuf/ptypes/timestamp"
	"github.com/container-storage-interface/spec/lib/go/csi"
	"github.com/dell/csi-unity/service/utils"
	"github.com/dell/gounity"
	"github.com/dell/gounity/types"
	"golang.org/x/net/context"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)
const (
	keyStoragePool          = "storagePool"
	keyThinProvisioned      = "thinProvisioned"
	keyDescription          = "description"
	keyDataReductionEnabled = "isDataReductionEnabled"
	keyTieringPolicy        = "tieringPolicy"
	keyHostIOLimitName      = "hostIOLimitName"
	keyArrayId              = "arrayId"
	keyProtocol             = "protocol"
	keyNasServer            = "nasServer"
	keyHostIoSize           = "hostIoSize"
)
const (
	FC                       = "FC"
	ISCSI                    = "iSCSI"
	NFS                      = "NFS"
	ProtocolUnknown          = "Unknown"
	ProtocolNFS              = int(0)
	MAX_ENTRIES_SNAPSHOT     = 100
	MAX_ENTRIES_VOLUME       = 100
	NFSShareLocalPath        = "/"
	NFSShareNamePrefix       = "csishare-"
	AdditionalFilesystemSize = 1.5 * 1024 * 1024 * 1024
)
var (
	errUnknownAccessType      = "unknown access type is not Block or Mount"
	errUnknownAccessMode      = "access mode cannot be UNKNOWN"
	errIncompatibleAccessMode = "access mode should be single node reader or single node writer"
	errNoMultiNodeWriter      = "Multi-node with writer(s) only supported for block access type"
	errNoMultiNodeReader      = "Multi-node Reader access mode is only supported for block access type"
	errBlockReadOnly          = "Read Only Many access mode not supported for Block Volume"
	errBlockNFS               = "Block Volume Capability is not supported for NFS"
)
type CRParams struct {
	VolumeName      string
	Protocol        string
	StoragePool     string
	Desciption      string
	HostIOLimitName string
	Thin            bool
	DataReduction   bool
	Size            int64
	TieringPolicy   int64
	HostIoSize      int64
}
type resourceType string
const volumeType resourceType = "volume"
const snapshotType resourceType = "snapshot"
func (s *service) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing CreateVolume with args: %+v", *req)
	params := req.GetParameters()
	arrayID := strings.ToLower(strings.TrimSpace(params[keyArrayId]))
	if arrayID == "" {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "ArrayId cannot be empty"))
	}
	ctx, log = setArrayIdContext(ctx, arrayID)
	if err := s.requireProbe(ctx, arrayID); err != nil {
		return nil, err
	}
	unity, err := s.getUnityClient(ctx, arrayID)
	if err != nil {
		return nil, err
	}
	protocol, storagePool, size, tieringPolicy, hostIoSize, thin, dataReduction, err := ValidateCreateVolumeRequest(ctx, req)
	if err != nil {
		return nil, err
	}
	volName := req.GetName()
	accessibility := req.GetAccessibilityRequirements()
	preferredAccessibility := accessibility.GetPreferred()
	log.Infof("PREFERRED-->%+v", preferredAccessibility)
	desc := params[keyDescription]
	hostIOLimitName := strings.TrimSpace(params[keyHostIOLimitName])
	crParams := CRParams{
		VolumeName:      volName,
		Protocol:        protocol,
		StoragePool:     storagePool,
		Desciption:      desc,
		HostIOLimitName: hostIOLimitName,
		Thin:            thin,
		DataReduction:   dataReduction,
		Size:            size,
		TieringPolicy:   tieringPolicy,
		HostIoSize:      hostIoSize,
	}
	contentSource := req.GetVolumeContentSource()
	if contentSource != nil {
		volumeSource := contentSource.GetVolume()
		if volumeSource != nil {
			sourceVolID := volumeSource.VolumeId
			log.Debugf("Cloning Volume: %s", sourceVolID)
			resp, err := s.createVolumeClone(ctx, &crParams, sourceVolID, arrayID, contentSource, unity, preferredAccessibility)
			return resp, err
		}
		snapshotSource := contentSource.GetSnapshot()
		if snapshotSource != nil {
			snapshotID := snapshotSource.SnapshotId
			log.Debugf("Create Volume from Snapshot: %s", snapshotID)
			resp, err := s.createVolumeFromSnap(ctx, &crParams, snapshotID, arrayID, contentSource, unity, preferredAccessibility)
			return resp, err
		}
	}
	if protocol == NFS {
		nasServer, ok := params[keyNasServer]
		if !ok {
			return nil, status.Errorf(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "`%s` is a required parameter", keyNasServer))
		}
		size += AdditionalFilesystemSize
		fields := map[string]interface{}{
			"storagePool":   storagePool,
			"Accessibility": accessibility,
			"contentSource": contentSource,
			"thin":          thin,
			"dataReduction": dataReduction,
			"tieringPolicy": tieringPolicy,
			"protocol":      protocol,
			"nasServer":     nasServer,
			"hostIoSize":    hostIoSize,
		}
		log.WithFields(fields).Infof("Executing Create File System with following fields")
		fileAPI := gounity.NewFilesystem(unity)
		filesystem, _ := fileAPI.FindFilesystemByName(ctx, volName)
		if filesystem != nil {
			content := filesystem.FileContent
			if int64(content.SizeTotal) == size && content.NASServer.Id == nasServer && content.Pool.Id == storagePool {
				log.Info("Filesystem exists in the requested state with same size, NAS server and storage pool")
				filesystem.FileContent.SizeTotal -= AdditionalFilesystemSize
				return utils.GetVolumeResponseFromFilesystem(filesystem, arrayID, protocol, preferredAccessibility), nil
			} else {
				log.Info("'Filesystem name' already exists and size/NAS server/storage pool is different")
				return nil, status.Error(codes.AlreadyExists, utils.GetMessageWithRunID(rid, "'Filesystem name' already exists and size/NAS server/storage pool is different."))
			}
		}
		log.Debug("Filesystem does not exist, proceeding to create new filesystem")
		resp, err := fileAPI.CreateFilesystem(ctx, volName, storagePool, desc, nasServer, uint64(size), int(tieringPolicy), int(hostIoSize), ProtocolNFS, thin, dataReduction)
		if err != nil {
			log.Debugf("Filesystem create response:%v Error:%v", resp, err)
			return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Create Filesystem %s failed with error: %v", volName, err))
		}
		resp, err = fileAPI.FindFilesystemByName(ctx, volName)
		if err != nil {
			log.Debugf("Find Filesystem response: %v Error: %v", resp, err)
		}
		if resp != nil {
			resp.FileContent.SizeTotal -= AdditionalFilesystemSize
			filesystemResp := utils.GetVolumeResponseFromFilesystem(resp, arrayID, protocol, preferredAccessibility)
			return filesystemResp, nil
		}
	} else {
		fields := map[string]interface{}{
			"storagePool":     storagePool,
			"Accessibility":   accessibility,
			"contentSource":   contentSource,
			"thin":            thin,
			"dataReduction":   dataReduction,
			"tieringPolicy":   tieringPolicy,
			"protocol":        protocol,
			"hostIOLimitName": hostIOLimitName,
		}
		log.WithFields(fields).Infof("Executing CreateVolume with following fields")
		volumeAPI := gounity.NewVolume(unity)
		var hostIOLimit *types.IoLimitPolicy
		var hostIOLimitId string
		if hostIOLimitName != "" {
			hostIOLimit, err = volumeAPI.FindHostIOLimitByName(ctx, hostIOLimitName)
			if err != nil {
				return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "HostIOLimitName %s not found. Error: %v", hostIOLimitName, err))
			}
			hostIOLimitId = hostIOLimit.IoLimitPolicyContent.Id
		}
		vol, _ := volumeAPI.FindVolumeByName(ctx, volName)
		if vol != nil {
			content := vol.VolumeContent
			if int64(content.SizeTotal) == size {
				log.Info("Volume exists in the requested state with same size")
				return utils.GetVolumeResponseFromVolume(vol, arrayID, protocol, preferredAccessibility), nil
			} else {
				log.Info("'Volume name' already exists and size is different")
				return nil, status.Error(codes.AlreadyExists, utils.GetMessageWithRunID(rid, "'Volume name' already exists and size is different."))
			}
		}
		log.Debug("Volume does not exist, proceeding to create new volume")
		resp, err := volumeAPI.CreateLun(ctx, volName, storagePool, desc, uint64(size), int(tieringPolicy), hostIOLimitId, thin, dataReduction)
		if err != nil {
			return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Create Volume %s failed with error: %v", volName, err))
		}
		resp, err = volumeAPI.FindVolumeByName(ctx, volName)
		if resp != nil {
			volumeResp := utils.GetVolumeResponseFromVolume(resp, arrayID, protocol, preferredAccessibility)
			log.Debugf("CreateVolume successful for volid: [%s]", volumeResp.Volume.VolumeId)
			return volumeResp, nil
		}
	}
	return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Volume/Filesystem not found after create. %v", err))
}
func (s *service) DeleteVolume(
	ctx context.Context,
	req *csi.DeleteVolumeRequest) (
	*csi.DeleteVolumeResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing DeleteVolume with args: %+v", *req)
	var snapErr error
	volID, protocol, arrayId, unity, err := s.validateAndGetResourceDetails(ctx, req.GetVolumeId(), volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	deleteVolumeResp := &csi.DeleteVolumeResponse{}
	if protocol != NFS {
		var throwErr error
		err, throwErr = s.deleteBlockVolume(ctx, volID, unity)
		if throwErr != nil {
			return nil, throwErr
		}
	} else {
		var throwErr error
		err, snapErr, throwErr = s.deleteFilesystem(ctx, volID, unity)
		if throwErr != nil {
			return nil, throwErr
		}
	}
	if err == nil {
		log.Debugf("DeleteVolume successful for volid: [%s]", req.VolumeId)
		return deleteVolumeResp, nil
	} else if err == gounity.FilesystemNotFoundError || err == gounity.VolumeNotFoundError || snapErr == gounity.SnapshotNotFoundError {
		log.Debug("Volume not found on array")
		log.Debugf("DeleteVolume successful for volid: [%s]", req.VolumeId)
		return deleteVolumeResp, nil
	}
	return nil, status.Error(codes.FailedPrecondition, utils.GetMessageWithRunID(rid, "Delete Volume %s failed with error: %v", volID, err))
}
func (s *service) ControllerPublishVolume(
	ctx context.Context,
	req *csi.ControllerPublishVolumeRequest) (
	*csi.ControllerPublishVolumeResponse, error) {
	ctx, log, _ := GetRunidLog(ctx)
	log.Debugf("Executing ControllerPublishVolume with args: %+v", *req)
	volID, protocol, arrayID, unity, err := s.validateAndGetResourceDetails(ctx, req.GetVolumeId(), volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayID)
	if err := s.requireProbe(ctx, arrayID); err != nil {
		return nil, err
	}
	protocol, nodeID, err := ValidateControllerPublishRequest(ctx, req, protocol)
	if err != nil {
		return nil, err
	}
	hostNames := strings.Split(nodeID, ",")
	host, err := s.getHostId(ctx, arrayID, hostNames[0], hostNames[1])
	if err != nil {
		return nil, err
	}
	hostContent := host.HostContent
	hostID := hostContent.ID
	pinfo := make(map[string]string)
	pinfo["volumeContextId"] = req.GetVolumeId()
	pinfo["arrayId"] = arrayID
	pinfo["host"] = nodeID
	vc := req.GetVolumeCapability()
	am := vc.GetAccessMode()
	if protocol == FC || protocol == ISCSI {
		resp, err := s.exportVolume(ctx, protocol, volID, hostID, nodeID, arrayID, unity, pinfo, host, vc)
		return resp, err
	}
	resp, err := s.exportFilesystem(ctx, volID, hostID, nodeID, arrayID, unity, pinfo, am)
	return resp, err
}
func (s *service) ControllerUnpublishVolume(
	ctx context.Context,
	req *csi.ControllerUnpublishVolumeRequest) (
	*csi.ControllerUnpublishVolumeResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing ControllerUnpublishVolume with args: %+v", *req)
	volID, protocol, arrayID, unity, err := s.validateAndGetResourceDetails(ctx, req.GetVolumeId(), volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayID)
	if err := s.requireProbe(ctx, arrayID); err != nil {
		return nil, err
	}
	nodeID := req.GetNodeId()
	if nodeID == "" {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "Node ID is required"))
	}
	hostNames := strings.Split(nodeID, ",")
	host, err := s.getHostId(ctx, arrayID, hostNames[0], hostNames[1])
	if err != nil {
		return nil, err
	}
	hostContent := host.HostContent
	hostID := hostContent.ID
	if protocol != NFS {
		volumeAPI := gounity.NewVolume(unity)
		vol, err := volumeAPI.FindVolumeById(ctx, volID)
		if err != nil {
			if err == gounity.VolumeNotFoundError {
				log.Debugf("Volume %s not found on the array %s during Controller Unpublish. Hence considering the call to be idempotent", volID, arrayID)
				return &csi.ControllerUnpublishVolumeResponse{}, nil
			}
			return nil, status.Error(codes.Internal, utils.GetMessageWithRunID(rid, "%v", err))
		}
		content := vol.VolumeContent
		if len(content.HostAccessResponse) > 0 {
			hostIDList := make([]string, 0)
			for _, hostaccess := range content.HostAccessResponse {
				hostcontent := hostaccess.HostContent
				hostAccessID := hostcontent.ID
				if hostAccessID != hostID {
					hostIDList = append(hostIDList, hostAccessID)
				}
			}
			log.Debug("Removing Host access on Volume ", volID)
			log.Debug("List of host access that will be retained on the volume: ", hostIDList)
			err = volumeAPI.ModifyVolumeExport(ctx, volID, hostIDList)
			if err != nil {
				return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Unexport Volume Failed. %v", err))
			}
		} else {
			log.Info(fmt.Sprintf("The given Node %s does not have access on the given volume %s. Already in Unpublished state.", hostID, volID))
		}
		log.Debugf("ControllerUnpublishVolume successful for volid: [%s]", req.GetVolumeId())
		return &csi.ControllerUnpublishVolumeResponse{}, nil
	}
	err = s.unexportFilesystem(ctx, volID, hostID, nodeID, req.GetVolumeId(), arrayID, unity)
	if err != nil {
		return nil, err
	}
	return &csi.ControllerUnpublishVolumeResponse{}, nil
}
func (s *service) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing ValidateVolumeCapabilities with args: %+v", *req)
	volID, _, arrayId, unity, err := s.validateAndGetResourceDetails(ctx, req.GetVolumeId(), volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	volumeAPI := gounity.NewVolume(unity)
	_, err = volumeAPI.FindVolumeById(ctx, volID)
	if err != nil {
		return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Volume not found. Error: %v", err))
	}
	params := req.GetParameters()
	protocol, _ := params[keyProtocol]
	if protocol == "" {
		log.Errorf("Protocol is required to validate capabilities")
		return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Protocol is required to validate capabilities"))
	}
	vcs := req.GetVolumeCapabilities()
	supported, reason := valVolumeCaps(vcs, protocol)
	resp := &csi.ValidateVolumeCapabilitiesResponse{}
	if supported {
		confirmed := &csi.ValidateVolumeCapabilitiesResponse_Confirmed{}
		confirmed.VolumeCapabilities = vcs
		resp.Confirmed = confirmed
		return resp, nil
	} else {
		resp.Message = reason
		return resp, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Unsupported capability"))
	}
}
func (s *service) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
	return nil, status.Error(codes.Unimplemented, "Not implemented")
}
func (s *service) GetCapacity(
	ctx context.Context,
	req *csi.GetCapacityRequest) (
	*csi.GetCapacityResponse, error) {
	return nil, status.Error(codes.Unimplemented, "Not implemented")
}
func (s *service) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing CreateSnapshot with args: %+v", *req)
	if len(req.SourceVolumeId) == 0 {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "Storage Resource ID cannot be empty"))
	}
	var err error
	req.Name, err = util.ValidateResourceName(req.Name, api.MaxResourceNameLength)
	if err != nil {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "invalid snapshot name [%v]", err))
	}
	volId, protocol, arrayId, _, err := s.validateAndGetResourceDetails(ctx, req.SourceVolumeId, volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	snap, err := s.createIdempotentSnapshot(ctx, req.Name, volId, req.Parameters["description"], req.Parameters["retentionDuration"], protocol, arrayId, false)
	if err != nil {
		return nil, err
	}
	return utils.GetSnapshotResponseFromSnapshot(snap, protocol, arrayId), nil
}
func (s *service) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing DeleteSnapshot with args: %+v", *req)
	snapId, _, arrayId, unity, err := s.validateAndGetResourceDetails(ctx, req.SnapshotId, snapshotType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	snapApi := gounity.NewSnapshot(unity)
	snap, err := snapApi.FindSnapshotById(ctx, snapId)
	if err != nil {
		log.Info("Snapshot doesn't exists")
	}
	if snap != nil {
		err := snapApi.DeleteSnapshot(ctx, snapId)
		if err != nil {
			return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Delete Snapshot error: %v", err))
		}
	}
	delSnapResponse := &csi.DeleteSnapshotResponse{}
	log.Debugf("Delete snapshot successful [%s]", req.SnapshotId)
	return delSnapResponse, nil
}
func (s *service) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Infof("Executing ListSnapshot with args: %+v", *req)
	var (
		startToken int
		err        error
		maxEntries = int(req.MaxEntries)
	)
	snapId, protocol, arrayId, unity, err := s.validateAndGetResourceDetails(ctx, req.SnapshotId, snapshotType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	snapApi := gounity.NewSnapshot(unity)
	if maxEntries > MAX_ENTRIES_SNAPSHOT || maxEntries == 0 {
		maxEntries = MAX_ENTRIES_SNAPSHOT
	}
	if req.StartingToken != "" {
		i, err := strconv.ParseInt(req.StartingToken, 10, 64)
		if err != nil {
			return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "Unable to parse StartingToken: %v into uint32", req.StartingToken))
		}
		startToken = int(i)
	}
	snaps, nextToken, err := snapApi.ListSnapshots(ctx, startToken, maxEntries, "", snapId)
	if err != nil {
		return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Unable to get the snapshots: %v", err))
	}
	entries, err := s.getCSISnapshots(snaps, req.SourceVolumeId, protocol, arrayId)
	if err != nil {
		return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, err.Error()))
	}
	log.Debugf("ListSnapshot successful for snapid: [%s]", req.SnapshotId)
	return &csi.ListSnapshotsResponse{
		Entries:   entries,
		NextToken: strconv.Itoa(nextToken),
	}, nil
}
func (s *service) controllerProbe(ctx context.Context, arrayId string) error {
	return s.probe(ctx, "Controller", arrayId)
}
func (cs *service) ControllerGetCapabilities(ctx context.Context, req *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
	ctx, log, _ := GetRunidLog(ctx)
	log.Debugf("Executing ControllerGetCapabilities with args: %+v", *req)
	return &csi.ControllerGetCapabilitiesResponse{
		Capabilities: []*csi.ControllerServiceCapability{
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_GET_CAPACITY,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
					},
				},
			},
			&csi.ControllerServiceCapability{
				Type: &csi.ControllerServiceCapability_Rpc{
					Rpc: &csi.ControllerServiceCapability_RPC{
						Type: csi.ControllerServiceCapability_RPC_PUBLISH_READONLY,
					},
				},
			},
		},
	}, nil
}
func (s *service) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
	ctx, log, rid := GetRunidLog(ctx)
	log.Debugf("Executing ControllerExpandVolume with args: %+v", *req)
	if req.VolumeId == "" {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "volumeId is mandatory parameter"))
	}
	volId, protocol, arrayId, unity, err := s.validateAndGetResourceDetails(ctx, req.VolumeId, volumeType)
	if err != nil {
		return nil, err
	}
	ctx, log = setArrayIdContext(ctx, arrayId)
	if err := s.requireProbe(ctx, arrayId); err != nil {
		return nil, err
	}
	var capacity int64
	if cr := req.CapacityRange; cr != nil {
		if rb := cr.RequiredBytes; rb > 0 {
			capacity = rb
		}
		if lb := cr.LimitBytes; lb > 0 {
			capacity = lb
		}
	}
	if capacity <= 0 {
		return nil, status.Error(codes.InvalidArgument, utils.GetMessageWithRunID(rid, "Required bytes can not be 0 or less"))
	}
	expandVolumeResp := &csi.ControllerExpandVolumeResponse{
		CapacityBytes: capacity,
	}
	if protocol == NFS {
		capacity += AdditionalFilesystemSize
		filesystemApi := gounity.NewFilesystem(unity)
		filesystem, err := filesystemApi.FindFilesystemById(ctx, volId)
		if err != nil {
			snapshotApi := gounity.NewSnapshot(unity)
			_, err = snapshotApi.FindSnapshotById(ctx, volId)
			if err != nil {
				return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Find filesystem %s failed with error: %v", volId, err))
			}
			return nil, status.Error(codes.Unimplemented, utils.GetMessageWithRunID(rid, "Expand Volume not supported for cloned filesystems(snapshot on array)"))
		}
		if filesystem.FileContent.SizeTotal >= uint64(capacity) {
			log.Infof("New Filesystem size (%d) is same as existing Filesystem size. Ignoring expand volume operation.", filesystem.FileContent.SizeTotal)
			expandVolumeResp := &csi.ControllerExpandVolumeResponse{
				CapacityBytes: 0,
			}
			expandVolumeResp.NodeExpansionRequired = false
			return expandVolumeResp, nil
		}
		err = filesystemApi.ExpandFilesystem(ctx, volId, uint64(capacity))
		if err != nil {
			return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Expand filesystem failed with error: %v", err))
		}
		filesystem, err = filesystemApi.FindFilesystemById(ctx, volId)
		if err != nil {
			return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Find filesystem failed with error: %v", err))
		}
		expandVolumeResp.CapacityBytes = int64(filesystem.FileContent.SizeTotal) - AdditionalFilesystemSize
		expandVolumeResp.NodeExpansionRequired = false
		return expandVolumeResp, err
	} else {
		volumeApi := gounity.NewVolume(unity)
		volume, err := volumeApi.FindVolumeById(ctx, volId)
		if err != nil {
			return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Find volume failed with error: %v", err))
		}
		nodeExpansionRequired := false
		content := volume.VolumeContent
		if len(content.HostAccessResponse) >= 1 {  
			nodeExpansionRequired = true
		}
		if volume.VolumeContent.SizeTotal >= uint64(capacity) {
			log.Infof("New Volume size (%d) is same as existing Volume size. Ignoring expand volume operation.", volume.VolumeContent.SizeTotal)
			expandVolumeResp := &csi.ControllerExpandVolumeResponse{
				CapacityBytes: 0,
			}
			expandVolumeResp.NodeExpansionRequired = nodeExpansionRequired
			return expandVolumeResp, nil
		}
		err = volumeApi.ExpandVolume(ctx, volId, uint64(capacity))
		if err != nil {
			return nil, status.Error(codes.Unknown, utils.GetMessageWithRunID(rid, "Expand volume failed with error: %v", err))
		}
		volume, err = volumeApi.FindVolumeById(ctx, volId)
		if err != nil {
			return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Find volume failed with error: %v", err))
		}
		expandVolumeResp.CapacityBytes = int64(volume.VolumeContent.SizeTotal)
		expandVolumeResp.NodeExpansionRequired = nodeExpansionRequired
		return expandVolumeResp, err
	}
}
func (s *service) getCSIVolumes(volumes []types.Volume) ([]*csi.ListVolumesResponse_Entry, error) {
	entries := make([]*csi.ListVolumesResponse_Entry, len(volumes))
	for i, vol := range volumes {
		attributes := map[string]string{
			"Name":          vol.VolumeContent.Name,
			"Type":          strconv.Itoa(vol.VolumeContent.Type),
			"Wwn":           vol.VolumeContent.Wwn,
			"StoragePoolID": vol.VolumeContent.Pool.Id,
		}
		vi := &csi.Volume{
			VolumeId:      vol.VolumeContent.ResourceId,
			CapacityBytes: int64(vol.VolumeContent.SizeTotal),
			VolumeContext: attributes,
		}
		entries[i] = &csi.ListVolumesResponse_Entry{
			Volume: vi,
		}
	}
	return entries, nil
}
func (s *service) getCSISnapshots(snaps []types.Snapshot, volId, protocol, arrayId string) ([]*csi.ListSnapshotsResponse_Entry, error) {
	entries := make([]*csi.ListSnapshotsResponse_Entry, len(snaps))
	for i, snap := range snaps {
		isReady := false
		if snap.SnapshotContent.State == 2 {
			isReady = true
		}
		var timestamp *timestamp.Timestamp
		if !snap.SnapshotContent.CreationTime.IsZero() {
			timestamp, _ = ptypes.TimestampProto(snap.SnapshotContent.CreationTime)
		}
		snapId := fmt.Sprintf("%s-%s-%s-%s", snap.SnapshotContent.Name, protocol, arrayId, snap.SnapshotContent.ResourceId)
		size := snap.SnapshotContent.Size
		if protocol == NFS {
			size -= AdditionalFilesystemSize
		}
		vi := &csi.Snapshot{
			SizeBytes:      size,
			SnapshotId:     snapId,
			SourceVolumeId: volId,
			CreationTime:   timestamp,
			ReadyToUse:     isReady,
		}
		entries[i] = &csi.ListSnapshotsResponse_Entry{
			Snapshot: vi,
		}
	}
	return entries, nil
}
func (s *service) getFilesystemByResourceID(ctx context.Context, resourceID, arrayID string) (*types.Filesystem, error) {
	ctx, _, rid := GetRunidLog(ctx)
	unity, err := s.getUnityClient(ctx, arrayID)
	if err != nil {
		return nil, err
	}
	filesystemAPI := gounity.NewFilesystem(unity)
	filesystemID, err := filesystemAPI.GetFilesystemIdFromResId(ctx, resourceID)
	if err != nil {
		return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Storage resource: %s filesystem Id not found. Error: %v", resourceID, err))
	}
	sourceFilesystemResp, err := filesystemAPI.FindFilesystemById(ctx, filesystemID)
	if err != nil {
		return nil, status.Error(codes.NotFound, utils.GetMessageWithRunID(rid, "Filesystem: %s not found. Error: %v", filesystemID, err))
	}
	return sourceFilesystemResp, nil
}