controller.go

/*
Package csi is CSI driver interface for OSD
Copyright 2017 Portworx

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 csi

import (
"encoding/json"
"fmt"
"math"
"reflect"
"sort"

"github.com/portworx/kvdb"
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/api/resource"

"github.com/libopenstorage/openstorage/api"
"github.com/libopenstorage/openstorage/api/server/sdk"
"github.com/libopenstorage/openstorage/pkg/grpcutil"
"github.com/libopenstorage/openstorage/pkg/units"
"github.com/libopenstorage/openstorage/pkg/util"

 csi "github.com/container-storage-interface/spec/lib/go/csi"
"golang.org/x/net/context"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

const (
// These keys are for accessing PVC Metadata added from the external-provisioner
osdParameterPrefix="csi.openstorage.org/"
osdPvcNameKey=osdParameterPrefix+"pvc-name"
osdPvcNamespaceKey=osdParameterPrefix+"pvc-namespace"
osdPvcAnnotationsKey=osdParameterPrefix+"pvc-annotations"
osdPvcLabelsKey=osdParameterPrefix+"pvc-labels"

// These keys are for accessing Snapshot Metadata added from the external-provisioner
osdSnapshotLabelsTypeKey=osdParameterPrefix+"snapshot-type"
osdSnapshotCredentialIDKey=osdParameterPrefix+"credential-id"

// in-tree keys for name and namespace
intreePvcNameKey="pvc"
intreePvcNamespaceKey="namespace"

// CSI keys for PVC metadata
csiPVNameKey="csi.storage.k8s.io/pv/name"
csiPVCNameKey="csi.storage.k8s.io/pvc/name"
csiPVCNamespaceKey="csi.storage.k8s.io/pvc/namespace"

volumeCapabilityMessageMultinodeVolume="Volume is a multinode volume"
volumeCapabilityMessageNotMultinodeVolume="Volume is not a multinode volume"
volumeCapabilityMessageReadOnlyVolume="Volume is read only"
volumeCapabilityMessageNotReadOnlyVolume="Volume is not read only"
defaultCSIVolumeSize=uint64(units.GiB*1)

// driver type
DriverTypeLocal="local"
DriverTypeCloud="cloud"
)

// ControllerGetCapabilities is a CSI API functions which returns to the caller
// the capabilities of the OSD CSI driver.
func (s*OsdCsiServer) ControllerGetCapabilities(
ctx context.Context,
req*csi.ControllerGetCapabilitiesRequest,
) (*csi.ControllerGetCapabilitiesResponse, error) {

caps:= []csi.ControllerServiceCapability_RPC_Type{
// Creating and deleting volumes supported
csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,

// Cloning: creation of volumes from snapshots, supported
csi.ControllerServiceCapability_RPC_CLONE_VOLUME,

// Resizing volumes supported
csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,

// Creating and deleting snapshots
csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,

// Listing snapshots
csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,

// Get single volume
csi.ControllerServiceCapability_RPC_GET_VOLUME,

// Volume condition
csi.ControllerServiceCapability_RPC_VOLUME_CONDITION,

// GetCapacity
csi.ControllerServiceCapability_RPC_GET_CAPACITY,
 }

varserviceCapabilities []*csi.ControllerServiceCapability

for_, cap:=rangecaps {
serviceCapabilities=append(serviceCapabilities, &csi.ControllerServiceCapability{
Type: &csi.ControllerServiceCapability_Rpc{
Rpc: &csi.ControllerServiceCapability_RPC{
Type: cap,
 },
 },
 })
 }

return&csi.ControllerGetCapabilitiesResponse{
Capabilities: serviceCapabilities,
 }, nil
}

// ControllerPublishVolume is a CSI API implements the attachment of a volume
// on to a node.
func (s*OsdCsiServer) ControllerPublishVolume(
 context.Context,
*csi.ControllerPublishVolumeRequest,
) (*csi.ControllerPublishVolumeResponse, error) {
returnnil, status.Error(codes.Unimplemented, "This request is not supported")
}

// ControllerUnpublishVolume is a CSI API which implements the detaching of a volume
// onto a node.
func (s*OsdCsiServer) ControllerUnpublishVolume(
 context.Context,
*csi.ControllerUnpublishVolumeRequest,
) (*csi.ControllerUnpublishVolumeResponse, error) {
returnnil, status.Error(codes.Unimplemented, "This request is not supported")
}

// ControllerGetVolume is a CSI API which implements getting a single volume.
// This function skips auth and directly hits the driver as it is read-only
// and only exposed via the CSI unix domain socket. If a secrets field is added
// in csi.ControllerGetVolumeRequest, we can update this to hit the SDK and use auth.
func (s*OsdCsiServer) ControllerGetVolume(
ctx context.Context,
req*csi.ControllerGetVolumeRequest,
) (*csi.ControllerGetVolumeResponse, error) {

clogger.WithContext(ctx).Tracef("ControllerGetVolume request received. VolumeID: %s", req.GetVolumeId())

vol, err:=s.driverGetVolume(ctx, req.GetVolumeId())
iferr!=nil {
ifs, ok:=status.FromError(err); ok&&s.Code() ==codes.NotFound {
return&csi.ControllerGetVolumeResponse{
Volume: &csi.Volume{
VolumeId: req.GetVolumeId(),
 },
Status: &csi.ControllerGetVolumeResponse_VolumeStatus{
VolumeCondition: &csi.VolumeCondition{
Abnormal: true,
Message: fmt.Sprintf("Volume ID %s not found", req.GetVolumeId()),
 },
 },
 }, nil
 }
returnnil, err
 }

return&csi.ControllerGetVolumeResponse{
Volume: &csi.Volume{
CapacityBytes: int64(vol.Spec.Size),
VolumeId: vol.Id,
 },
Status: &csi.ControllerGetVolumeResponse_VolumeStatus{
VolumeCondition: getVolumeCondition(vol),
 },
 }, nil
}

// ValidateVolumeCapabilities is a CSI API used by container orchestration systems
// to make sure a volume specification is validiated by the CSI driver.
func (s*OsdCsiServer) ValidateVolumeCapabilities(
ctx context.Context,
req*csi.ValidateVolumeCapabilitiesRequest,
) (*csi.ValidateVolumeCapabilitiesResponse, error) {

capabilities:=req.GetVolumeCapabilities()
ifcapabilities==nil||len(capabilities) ==0 {
returnnil, status.Error(codes.InvalidArgument, "volume_capabilities must be specified")
 }
id:=req.GetVolumeId()
iflen(id) ==0 {
returnnil, status.Error(codes.InvalidArgument, "volume_id must be specified")
 }

// Log request
clogger.WithContext(ctx).Infof("csi.ValidateVolumeCapabilities of id %s "+
"capabilities %#v ", id, capabilities)

// Get grpc connection
conn, err:=s.getConn()
iferr!=nil {
returnnil, status.Errorf(
codes.Unavailable,
"Unable to connect to SDK server: %v", err)
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)
defercancel()

// Check ID is valid with the specified volume capabilities
volumes:=api.NewOpenStorageVolumeClient(conn)
resp, err:=volumes.Inspect(ctx, &api.SdkVolumeInspectRequest{
VolumeId: id,
 })
iferr!=nil {
returnnil, status.Error(codes.NotFound, "ID not found")
 }
v:=resp.GetVolume()
ifv.Id!=id {
errs:=fmt.Sprintf(
"Driver volume id [%s] does not equal requested id of: %s",
v.Id,
id)
clogger.WithContext(ctx).Errorln(errs)
returnnil, status.Error(codes.Internal, errs)
 }
// Setup uninitialized response object
result:=&csi.ValidateVolumeCapabilitiesResponse{
Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{
VolumeContext: req.GetVolumeContext(),
VolumeCapabilities: req.GetVolumeCapabilities(),
Parameters: req.GetParameters(),
 },
 }

// Check capability
for_, capability:=rangecapabilities {
// Currently the CSI spec defines all storage as "file systems."
// So we do not need to check this with the volume. All we will check
// here is the validity of the capability access type.
ifcapability.GetMount() ==nil&&capability.GetBlock() ==nil {
returnnil, status.Error(
codes.InvalidArgument,
"Cannot have both mount and block be undefined")
 }

// Check access mode is setup correctly
mode:=capability.GetAccessMode()
switch {
casemode.Mode==csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER:
ifv.Spec.Sharedv4||v.Spec.Shared {
result.Confirmed=nil
result.Message=volumeCapabilityMessageMultinodeVolume
break
 }
ifv.Readonly {
result.Confirmed=nil
result.Message=volumeCapabilityMessageReadOnlyVolume
break
 }
casemode.Mode==csi.VolumeCapability_AccessMode_SINGLE_NODE_READER_ONLY:
ifv.Spec.Sharedv4||v.Spec.Shared {
result.Confirmed=nil
result.Message=volumeCapabilityMessageMultinodeVolume
break
 }
if!v.Readonly {
result.Confirmed=nil
result.Message=volumeCapabilityMessageNotReadOnlyVolume
break
 }
casemode.Mode==csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY:
if!v.Spec.Sharedv4&&!v.Spec.Shared {
result.Confirmed=nil
result.Message=volumeCapabilityMessageNotMultinodeVolume
break
 }
if!v.Readonly {
result.Confirmed=nil
result.Message=volumeCapabilityMessageNotReadOnlyVolume
break
 }
casemode.Mode==csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER||
mode.Mode==csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER:
if!v.Spec.Sharedv4&&!v.Spec.Shared {
result.Confirmed=nil
result.Message=volumeCapabilityMessageNotMultinodeVolume
break
 }
ifv.Readonly {
result.Confirmed=nil
result.Message=volumeCapabilityMessageReadOnlyVolume
break
 }
default:
returnnil, status.Errorf(
codes.InvalidArgument,
"AccessMode %s is not allowed",
mode.Mode.String())
 }

ifresult.Confirmed==nil {
returnresult, nil
 }
 }

// If we passed all the checks, then it is valid.
// result.Message needs to be empty on return
returnresult, nil
}

// osdVolumeContext returns the attributes of a volume as a map
// to be returned to the CSI API caller
funcosdVolumeContext(v*api.Volume) map[string]string {
returnmap[string]string{
api.SpecParent: v.GetSource().GetParent(),
api.SpecSecure: fmt.Sprintf("%v", v.GetSpec().GetEncrypted()),
api.SpecShared: fmt.Sprintf("%v", v.GetSpec().GetShared()),
api.SpecSharedv4: fmt.Sprintf("%v", v.GetSpec().GetSharedv4()),
"readonly": fmt.Sprintf("%v", v.GetReadonly()),
"attached": v.AttachedState.String(),
"state": v.State.String(),
"error": v.GetError(),
 }
}

funcaddJsonMapToMetadata(paramsstring, metadatamap[string]string) (map[string]string, error) {
decodedParams:=make(map[string]string)
iflen(params) >0 {
// Decode and add labels
err:=json.Unmarshal([]byte(params), &decodedParams)
iferr!=nil {
returnmetadata, err
 }
fork, v:=rangedecodedParams {
metadata[k] =v
 }
 }

returnmetadata, nil
}

funcgetPVCMetadata(paramsmap[string]string) (map[string]string, error) {
metadata:=make(map[string]string)

// Only add PVC name/namespace if present. Do not add empty values.
// This is useful for non-k8s schedulers.
ifpvcName, ok:=params[osdPvcNameKey]; ok {
metadata[intreePvcNameKey] =pvcName
 }
ifpvcNamespace, ok:=params[osdPvcNamespaceKey]; ok {
metadata[intreePvcNamespaceKey] =pvcNamespace
 }
metadata, err:=addJsonMapToMetadata(params[osdPvcAnnotationsKey], metadata)
iferr!=nil {
returnnil, err
 }
metadata, err=addJsonMapToMetadata(params[osdPvcLabelsKey], metadata)
iferr!=nil {
returnnil, err
 }
ifpvcName, ok:=params[csiPVCNameKey]; ok {
metadata[intreePvcNameKey] =pvcName
 }
ifpvcNamespace, ok:=params[csiPVCNamespaceKey]; ok {
metadata[intreePvcNamespaceKey] =pvcNamespace
 }

returnmetadata, nil
}

funccleanupVolumeLabels(labelsmap[string]string) map[string]string {
delete(labels, osdPvcNameKey)
delete(labels, osdPvcNamespaceKey)
delete(labels, osdPvcAnnotationsKey)
delete(labels, osdPvcLabelsKey)
delete(labels, csiPVNameKey)
delete(labels, csiPVCNameKey)
delete(labels, csiPVCNamespaceKey)

returnlabels
}

funcvalidateCreateVolumeCapabilities(caps []*csi.VolumeCapability) error {
iflen(caps) ==0 {
returnstatus.Error(codes.InvalidArgument, "Volume capabilities must be provided")
 }

varsharedbool
varblockbool
for_, cap:=rangecaps {
mode:=cap.GetAccessMode().GetMode()
ifmode==csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY {
shared=true
 }

ifcap.GetBlock() !=nil {
block=true
 }
 }

ifblock&&shared {
returnstatus.Errorf(
codes.InvalidArgument,
"Shared raw block volumes are not supported")
 }

returnnil
}

funcvalidateCreateVolumeCapabilitiesPure(caps []*csi.VolumeCapability, proxySpec*api.ProxySpec) error {
iflen(caps) ==0 {
returnstatus.Error(codes.InvalidArgument, "Volume capabilities must be provided")
 }

varsharedbool
varmountbool
varblockbool
for_, cap:=rangecaps {
mode:=cap.GetAccessMode().GetMode()
ifmode==csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY {
shared=true
 }

ifcap.GetBlock() !=nil {
block=true
 }

ifcap.GetMount() !=nil {
mount=true
 }
 }

// Check for shared FA DA volumes. Shared filesystems aren't supported.
// Shared raw block volumes are temporarily disabled due to PWX-23530.
// All FA raw block volumes are disabled for now.
ifproxySpec.ProxyProtocol==api.ProxyProtocol_PROXY_PROTOCOL_PURE_BLOCK {
ifmount&&shared {
returnstatus.Errorf(
codes.InvalidArgument,
"FlashArray Direct Access shared filesystems are not supported",
 )
 }
 }

// Check for FB DA volumes: all allowed except raw block
ifproxySpec.ProxyProtocol==api.ProxyProtocol_PROXY_PROTOCOL_PURE_FILE&&block {
returnstatus.Errorf(
codes.InvalidArgument,
"FlashBlade Direct Access volumes do not support raw block",
 )
 }

returnnil
}

// CreateVolume is a CSI API which creates a volume on OSD
// This function supports snapshots if the parent volume id is supplied
// in the parameters.
func (s*OsdCsiServer) CreateVolume(
ctx context.Context,
req*csi.CreateVolumeRequest,
) (*csi.CreateVolumeResponse, error) {
// Log request
clogger.WithContext(ctx).Infof("csi.CreateVolume request received. Volume: %s", req.GetName())

iflen(req.GetName()) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Name must be provided")
 }

// Get parameters
spec, locator, source, err:=s.specHandler.SpecFromOpts(req.GetParameters())
iferr!=nil {
e:=fmt.Sprintf("Unable to get parameters: %s\n", err.Error())
clogger.WithContext(ctx).Errorln(e)
returnnil, status.Error(codes.InvalidArgument, e)
 }

// Check ID is valid with the specified volume capabilities
snapshotType, ok:=locator.VolumeLabels[osdSnapshotLabelsTypeKey]
if!ok {
snapshotType=DriverTypeLocal
 }
ifDriverTypeCloud==snapshotType&&req.VolumeContentSource!=nil&&req.VolumeContentSource.GetSnapshot() !=nil {
clogger.WithContext(ctx).Infof("csi.CreateVolume restoring snapshot to Volume: %s", req.GetName())
returns.restoreSnapshot(ctx, req)
 }

ifspec.IsPureVolume() {
err=validateCreateVolumeCapabilitiesPure(req.GetVolumeCapabilities(), spec.GetProxySpec())
iferr!=nil {
returnnil, err
 }
 } else {
err=validateCreateVolumeCapabilities(req.GetVolumeCapabilities())
iferr!=nil {
returnnil, err
 }
 }

// Get PVC Metadata and add to locator.VolumeLabels
// This will override any storage class secrets added above.
pvcMetadata, err:=getPVCMetadata(req.GetParameters())
iferr!=nil {
returnnil, status.Errorf(
codes.Internal,
"Unable to get PVC Metadata: %v", err)
 }
fork, v:=rangepvcMetadata {
locator.VolumeLabels[k] =v
 }

// Copy all SC Parameters (from req.Parameters) to locator.VolumeLabels.
// This explicit copy matches the equivalent behavior in the in-tree driver
iflen(locator.VolumeLabels) ==0 {
locator.VolumeLabels=make(map[string]string)
 }
fork, v:=rangereq.Parameters {
locator.VolumeLabels[k] =v
 }

// Add encryption secret information to VolumeLabels
locator.VolumeLabels=s.addEncryptionInfoToLabels(locator.VolumeLabels, req.GetSecrets())

// Get parent ID from request: snapshot or volume
ifreq.GetVolumeContentSource() !=nil {
ifsourceSnap:=req.GetVolumeContentSource().GetSnapshot(); sourceSnap!=nil {
source.Parent=sourceSnap.SnapshotId
 }

ifsourceVol:=req.GetVolumeContentSource().GetVolume(); sourceVol!=nil {
source.Parent=sourceVol.VolumeId
 }
 }

// Get Size
ifreq.GetCapacityRange() !=nil&&req.GetCapacityRange().GetRequiredBytes() !=0 {
size, err:=roundUpToNearestGiB(req.GetCapacityRange().GetRequiredBytes())
iferr!=nil {
returnnil, fmt.Errorf("failed to round volume size up to nearest GiB: %v", err.Error())
 }
spec.Size=size

 } else {
spec.Size=defaultCSIVolumeSize
 }

// cleanup duplicate information after pulling from req.GetParameters
locator.VolumeLabels=cleanupVolumeLabels(locator.VolumeLabels)

// Get grpc connection
conn, err:=s.getRemoteConn(ctx)
iferr!=nil {
logrus.Errorf("failed to get remote connection: %v, continuing with local node instead", err)
conn, err=s.getConn()
iferr!=nil {
returnnil, status.Errorf(
codes.Unavailable,
"Unable to connect to SDK server: %v", err)
 }
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)
defercancel()

// Check ID is valid with the specified volume capabilities
volumes:=api.NewOpenStorageVolumeClient(conn)

// Create volume
varnewVolumeIdstring
ifsource.Parent=="" {
spec, err:=resolveSpecFromCSI(spec, req)
iferr!=nil {
returnnil, err
 }

varcreateResp*api.SdkVolumeCreateResponse
varcreateErrerror
topologies:=getAllTopologies(req.AccessibilityRequirements)
ifspec.IsPureVolume() &&len(topologies) >0 {
for_, topo:=rangetopologies {
spec.TopologyRequirement=&api.TopologyRequirement{
Labels: topo.Segments,
 }
createResp, createErr=volumes.Create(ctx, &api.SdkVolumeCreateRequest{
Name: req.GetName(),
Spec: spec,
Labels: locator.GetVolumeLabels(),
 })
ifs, ok:=status.FromError(createErr); createErr==nil|| (ok&&s.Code() !=codes.ResourceExhausted) {
break
 }
 }
 } else {
createResp, createErr=volumes.Create(ctx, &api.SdkVolumeCreateRequest{
Name: req.GetName(),
Spec: spec,
Labels: locator.GetVolumeLabels(),
 })
 }
ifcreateErr!=nil {
returnnil, createErr
 }
newVolumeId=createResp.VolumeId
 } else {
labels:=locator.GetVolumeLabels()
ifspec.GetFADAPodName() !="" {
labels[api.SpecPurePodName] =spec.GetFADAPodName()
 }
cloneResp, err:=volumes.Clone(ctx, &api.SdkVolumeCloneRequest{
Name: req.GetName(),
ParentId: source.Parent,
AdditionalLabels: labels,
 })
iferr!=nil {
returnnil, err
 }

newVolumeId=cloneResp.VolumeId
 }

// Get volume information
inspectResp, err:=volumes.Inspect(ctx, &api.SdkVolumeInspectRequest{
VolumeId: newVolumeId,
 })
iferr!=nil {
returnnil, err
 }

// Create response
volume:=&csi.Volume{}
osdToCsiVolumeInfo(volume, inspectResp.GetVolume(), req)
return&csi.CreateVolumeResponse{
Volume: volume,
 }, nil
}

func (s*OsdCsiServer) restoreSnapshot(ctx context.Context,
req*csi.CreateVolumeRequest,
) (resp*csi.CreateVolumeResponse, errerror) {
clogger.WithContext(ctx).Infof("csi.CreateVolume is restoring snapshot. Volume: %s Snapshot: %s", req.GetName(), req.VolumeContentSource.GetSnapshot())
snapshot:=req.VolumeContentSource.GetSnapshot()
ifsnapshot==nil {
returnnil, status.Error(codes.NotFound, "snapshot fetched is not accurate or does not exist")
 }

cloudBackupClient, err:=s.getCloudBackupClient(ctx)
cloudBackupClientAvailable:=cloudBackupClient!=nil
cloudBackupDriverDisabled:=sdk.IsErrorUnavailable(err)
if (err!=nil&&!cloudBackupDriverDisabled) ||cloudBackupClient==nil {
returnnil, err
 }

csiSnapshotID:=snapshot.GetSnapshotId()
iflen(csiSnapshotID) ==0 {
returnnil, status.Error(codes.InvalidArgument, "snapshot id must be provided")
 }

varbackupStatus*api.SdkCloudBackupStatusResponse
ifcloudBackupClientAvailable&&!cloudBackupDriverDisabled {
// Check if snapshot has been created but is in error state
backupStatus, err=cloudBackupClient.Status(ctx, &api.SdkCloudBackupStatusRequest{
TaskId: csiSnapshotID,
 })
 }

isSnapshotIDPresentInCloud:=true
ifbackupStatus!=nil {
_, isSnapshotIDPresentInCloud=backupStatus.Statuses[csiSnapshotID]
 }

if (sdk.IsErrorNotFound(err) &&!cloudBackupDriverDisabled&&cloudBackupClientAvailable) ||!isSnapshotIDPresentInCloud {
clogger.WithContext(ctx).Infof("csi.CreateVolume is restoring snapshot. Volume: %s Snapshot: %s is a local backup", req.GetName(), csiSnapshotID)
return
 }
clogger.WithContext(ctx).Infof("csi.CreateVolume is restoring snapshot. Volume: %s Snapshot: %s is a cloud backup", req.GetName(), csiSnapshotID)

resp, err=s.restoreCloudSnapshot(ctx, req)
return

}

func (s*OsdCsiServer) restoreCloudSnapshot(ctx context.Context,
req*csi.CreateVolumeRequest,
) (resp*csi.CreateVolumeResponse, errerror) {
cloudBackupClient, err:=s.getCloudBackupClient(ctx)
iferr!=nil {
returnnil, err
 }
// Get parameters
_, locator, _, _:=s.specHandler.SpecFromOpts(req.GetParameters())

csiSnapshotID:=req.VolumeContentSource.GetSnapshot().GetSnapshotId()

clogger.WithContext(ctx).Infof("csi.CreateVolume is restoring snapshot. Volume: %s Snapshot: %s is a cloud backup with labels %+v", req.GetName(), csiSnapshotID, locator.VolumeLabels)

credentialID, ok:=locator.VolumeLabels[osdSnapshotCredentialIDKey]
if!ok {
e:=fmt.Sprintf("csi.CreateVolume is restoring snapshot. Volume: %s Snapshot: %s credentials missing", req.GetName(), csiSnapshotID)
clogger.WithContext(ctx).Infof(e)
returnnil, status.Error(codes.InvalidArgument, e)
 }

snapResp, err:=cloudBackupClient.Restore(ctx, &api.SdkCloudBackupRestoreRequest{
BackupId: csiSnapshotID,
RestoreVolumeName: req.GetName(),
TaskId: req.GetName(),
Locator: locator,
CredentialId: credentialID,
 })
ifnil!=err {
returnnil, status.Error(codes.Internal, err.Error())
 }

resp=&csi.CreateVolumeResponse{
Volume: &csi.Volume{
VolumeId: snapResp.GetRestoreVolumeId(),
ContentSource: req.VolumeContentSource,
 },
 }
return
}

funcgetClonedPVCMetadata(locator*api.VolumeLocator) map[string]string {
metadataLabels:=map[string]string{}
pvcName, ok:=locator.VolumeLabels[intreePvcNameKey]
ifok {
metadataLabels[intreePvcNameKey] =pvcName
 }
pvcNamespace, ok:=locator.VolumeLabels[intreePvcNamespaceKey]
ifok {
metadataLabels[intreePvcNamespaceKey] =pvcNamespace
 }

returnmetadataLabels
}

// DeleteVolume is a CSI API which deletes a volume
func (s*OsdCsiServer) DeleteVolume(
ctx context.Context,
req*csi.DeleteVolumeRequest,
) (*csi.DeleteVolumeResponse, error) {
// Log request
clogger.WithContext(ctx).Infof("csi.DeleteVolume request received. VolumeID: %s", req.VolumeId)

// Check arguments
iflen(req.GetVolumeId()) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 }

// Get grpc connection
conn, err:=s.getRemoteConn(ctx)
iferr!=nil {
logrus.Errorf("failed to get remote connection: %v, continuing with local node instead", err)
conn, err=s.getConn()
iferr!=nil {
returnnil, status.Errorf(
codes.Unavailable,
"Unable to connect to SDK server: %v", err)
 }
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)
defercancel()

// Check ID is valid with the specified volume capabilities
volumes:=api.NewOpenStorageVolumeClient(conn)

// Delete volume
if_, err=volumes.Delete(ctx, &api.SdkVolumeDeleteRequest{
VolumeId: req.GetVolumeId(),
 }); err!=nil {
e:=fmt.Sprintf("Unable to delete volume with id %s: %s",
req.GetVolumeId(),
err.Error())
clogger.WithContext(ctx).Errorln(e)
returnnil, status.Error(codes.Aborted, e)
 }

return&csi.DeleteVolumeResponse{}, nil
}

// ControllerExpandVolume is a CSI API which resizes a volume
func (s*OsdCsiServer) ControllerExpandVolume(
ctx context.Context,
req*csi.ControllerExpandVolumeRequest,
) (*csi.ControllerExpandVolumeResponse, error) {
iflen(req.GetVolumeId()) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 } elseifreq.GetCapacityRange() ==nil {
returnnil, status.Error(codes.InvalidArgument, "Capacity range must be provided")
 } elseifreq.GetCapacityRange().GetRequiredBytes() <0||req.GetCapacityRange().GetLimitBytes() <0 {
returnnil, status.Error(codes.InvalidArgument, "Capacity ranges values cannot be negative")
 }

// Get Size
spec:=&api.VolumeSpecUpdate{}
newSize, err:=roundUpToNearestGiB(req.GetCapacityRange().GetRequiredBytes())
iferr!=nil {
returnnil, fmt.Errorf("failed to round volume size up to nearest GiB: %v", err.Error())
 }
spec.SizeOpt=&api.VolumeSpecUpdate_Size{
Size: newSize,
 }

// Get grpc connection
conn, err:=s.getConn()
iferr!=nil {
returnnil, status.Errorf(
codes.Unavailable,
"Unable to connect to SDK server: %v", err)
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)
defercancel()

// If the new size is greater than the current size, a volume update
// should be issued. Otherwise, no operation should occur.
volumes:=api.NewOpenStorageVolumeClient(conn)

// Check for idempotency. There are cases where we return an error in the proto
// driver if the volume update request has no changes, so we return early here instead.
inspectResp, err:=volumes.Inspect(ctx, &api.SdkVolumeInspectRequest{
VolumeId: req.GetVolumeId(),
 })
iferr!=nil {
iferr==kvdb.ErrNotFound {
returnnil, status.Errorf(codes.NotFound, "Volume id %s not found", req.GetVolumeId())
 }
returnnil, status.Errorf(codes.Internal, "Failed to get volume %s: %v", req.GetVolumeId(), err)
 }
ifinspectResp.Volume.Spec.Size==spec.GetSize() {
return&csi.ControllerExpandVolumeResponse{
CapacityBytes: int64(newSize),
NodeExpansionRequired: false,
 }, nil
 }

// Update volume with new size
_, err=volumes.Update(ctx, &api.SdkVolumeUpdateRequest{
VolumeId: req.GetVolumeId(),
Spec: spec,
 })
iferr!=nil {
iferr==kvdb.ErrNotFound {
returnnil, status.Errorf(codes.NotFound, "Volume id %s not found", req.GetVolumeId())
 }
returnnil, status.Errorf(codes.Internal, "Failed to update volume size: %v", err)
 }

return&csi.ControllerExpandVolumeResponse{
CapacityBytes: int64(newSize),
NodeExpansionRequired: false,
 }, nil
}

funcosdToCsiVolumeInfo(dest*csi.Volume, src*api.Volume, req*csi.CreateVolumeRequest) {
dest.VolumeId=src.GetId()
dest.CapacityBytes=int64(src.Spec.GetSize())
dest.VolumeContext=osdVolumeContext(src)
dest.ContentSource=req.GetVolumeContentSource()
ifsrc.Spec.GetTopologyRequirement() !=nil&&len(src.Spec.TopologyRequirement.Labels) >0 {
dest.AccessibleTopology=make([]*csi.Topology, 1)
segments:=map[string]string{}
fork, v:=rangesrc.Spec.TopologyRequirement.Labels {
segments[k] =v
 }
dest.AccessibleTopology[0] =&csi.Topology{
Segments: segments,
 }
 }
}

// isFilesystemSpecSet checks if the fs parameter has been declared
funcisFilesystemSpecSet(paramsmap[string]string) bool {
_, fsSet:=params[api.SpecFilesystem]
returnfsSet
}

// resolveSharedSpec makes the following assumptions:
// 1. When a volume is set to RWX or a similar multi-node access mode, we default to Sharedv4
// 2. If a user prefers shared over sharedv4, they may still use it by explicitly declaring "shared": true
funcresolveSharedSpec(spec*api.VolumeSpec, req*csi.CreateVolumeRequest) (*api.VolumeSpec, error) {
// shared or sharedv4 parameter doesn't apply to pure backends so don't set them
ifspec.IsPureVolume() {
returnspec, nil
 }

// don't default to sharedv4/shared for RWX Volumes if proxy spec is set
ifspec.ProxySpec!=nil {
returnspec, nil
 }

varsharedbool
for_, cap:=rangereq.GetVolumeCapabilities() {
mode:=cap.GetAccessMode().GetMode()
ifmode==csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER||
mode==csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY {
shared=true
 }
 }

// Handle legacy support for shared
ifspec.Shared&&!spec.Sharedv4 {
spec.Shared=shared
 } else {
// assume sharedv4 by default
spec.Sharedv4=shared
 }

returnspec, nil
}

// resolveFSTypeSpec makes the following assumptions:
// 1. If provided, the PX "fstype" parameter should always override corresponding the CSI parameter.
// 2. The default value for spec.Format is determined upstream by SpecFromOpts(req.GetParameters())
funcresolveFSTypeSpec(spec*api.VolumeSpec, req*csi.CreateVolumeRequest) (*api.VolumeSpec, error) {
varcsiFsTypestring
for_, cap:=rangereq.GetVolumeCapabilities() {
ifcap.GetBlock() !=nil {
spec.Format=api.FSType_FS_TYPE_NONE
returnspec, nil
 }

// Get FsType according to CSI spec
ifmount:=cap.GetMount(); mount!=nil {
csiFsType=mount.FsType
 }
 }

// if we have the FileSystemSpec option not set, this means the user didn't intend to
// set the filesystem type based on the "PX way". In this case, we can safely
// apply the fsType from the CSI request.
if!isFilesystemSpecSet(req.GetParameters()) &&csiFsType!="" {
format, err:=api.FSTypeSimpleValueOf(csiFsType)
iferr!=nil {
returnspec, err
 }
spec.Format=format
 }

returnspec, nil
}

// resolveSpecFromCSI alters the api.VolumeSpec based on any CSI parameters passed in.
// Various volume spec fields have CSI equivalents. This function resolves each one.
funcresolveSpecFromCSI(spec*api.VolumeSpec, req*csi.CreateVolumeRequest) (*api.VolumeSpec, error) {
// Handles shared vs. sharedv4 resolution. We default to Sharedv4
spec, err:=resolveSharedSpec(spec, req)
iferr!=nil {
returnnil, err
 }

// Handle FsType resolution. There are two methods of setting the FsType: PX and CSI.
// We default to the PX parameter if it is provided.
// If CSI parameter is provided, but PX is not, use CSI.
// If neither is provided, use the PX parameter default.
spec, err=resolveFSTypeSpec(spec, req)
iferr!=nil {
returnnil, err
 }

returnspec, nil
}

funcisSnapshotReady(v*api.Volume) bool {
returnv.GetError() ==""&&v.Status==api.VolumeStatus_VOLUME_STATUS_UP
}

// CreateSnapshot is a CSI implementation to create a snapshot from the volume
func (s*OsdCsiServer) CreateSnapshot(
ctx context.Context,
req*csi.CreateSnapshotRequest,
) (*csi.CreateSnapshotResponse, error) {

iflen(req.GetSourceVolumeId()) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 } elseiflen(req.GetName()) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Name must be provided")
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)