// NodePublishVolume is a CSI API call which mounts the volume on the specified
// target path on the node.
//
// TODO: Support READ ONLY Mounts
//
491909
func (s *OsdCsiServer) NodePublishVolume(
	ctx context.Context,
	req *csi.NodePublishVolumeRequest,
) (*csi.NodePublishVolumeResponse, error) {

	logrus.Debugf("NodePublishVolume req[%#v]", req)

	// Check arguments
	if len(req.GetVolumeId()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "Volume id must be provided")
	}
	if len(req.GetTargetPath()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "Target path must be provided")
	}
	if req.GetVolumeCapability() == nil || req.GetVolumeCapability().GetAccessMode() == nil {
		return nil, status.Error(codes.InvalidArgument, "Volume access mode must be provided")
	}

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

	// Get secret if any was passed
	ctx = s.setupContextWithToken(ctx, req.GetSecrets())

	// Check if block device
	driverType := s.driver.Type()
	if driverType != api.DriverType_DRIVER_TYPE_BLOCK &&
		req.GetVolumeCapability().GetBlock() != nil {
		return nil, status.Errorf(codes.InvalidArgument, "Trying to attach as block a non block device")
	}

	// Gather volume attributes
	spec, _, _, err := s.specHandler.SpecFromOpts(req.GetVolumeContext())
	if err != nil {
		return nil, status.Errorf(
			codes.InvalidArgument,
			"Invalid volume attributes: %#v",
			req.GetVolumeContext())
	}

	// prepare for mount/attaching
	mounts := api.NewOpenStorageMountAttachClient(conn)
	opts := &api.SdkVolumeAttachOptions{
		SecretName: spec.GetPassphrase(),
	}
	if driverType == api.DriverType_DRIVER_TYPE_BLOCK {
		if _, err = mounts.Attach(ctx, &api.SdkVolumeAttachRequest{
			VolumeId: req.GetVolumeId(),
			Options:  opts,
		}); err != nil {
			return nil, err
		}
	}

	// Verify target location is an existing directory
	if err := verifyTargetLocation(req.GetTargetPath()); err != nil {
		return nil, status.Errorf(
			codes.Aborted,
			"Failed to use target location %s: %s",
			req.GetTargetPath(),
			err.Error())
	}

	// Mount volume onto the path
	if _, err := mounts.Mount(ctx, &api.SdkVolumeMountRequest{
		VolumeId:  req.GetVolumeId(),
		MountPath: req.GetTargetPath(),
		Options:   opts,
	}); err != nil {
		return nil, err
	}

	logrus.Infof("Volume %s mounted on %s",
		req.GetVolumeId(),
		req.GetTargetPath())

	return &csi.NodePublishVolumeResponse{}, nil
}
239360
func (m *csiBlockMapper) stageVolumeForBlock(
	ctx context.Context,
	csi csiClient,
	accessMode v1.PersistentVolumeAccessMode,
	csiSource *v1.CSIPersistentVolumeSource,
	attachment *storage.VolumeAttachment,
) (string, error) {
	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock called"))

	stagingPath := m.getStagingPath()
	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock stagingPath set [%s]", stagingPath))

	// Check whether "STAGE_UNSTAGE_VOLUME" is set
	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
	if err != nil {
		klog.Error(log("blockMapper.stageVolumeForBlock failed to check STAGE_UNSTAGE_VOLUME capability: %v", err))
		return "", err
	}
	if !stageUnstageSet {
		klog.Infof(log("blockMapper.stageVolumeForBlock STAGE_UNSTAGE_VOLUME capability not set. Skipping MountDevice..."))
		return "", nil
	}

	publishVolumeInfo := attachment.Status.AttachmentMetadata

	nodeStageSecrets := map[string]string{}
	if csiSource.NodeStageSecretRef != nil {
		nodeStageSecrets, err = getCredentialsFromSecret(m.k8s, csiSource.NodeStageSecretRef)
		if err != nil {
			return "", fmt.Errorf("failed to get NodeStageSecretRef %s/%s: %v",
				csiSource.NodeStageSecretRef.Namespace, csiSource.NodeStageSecretRef.Name, err)
		}
	}

	// Creating a stagingPath directory before call to NodeStageVolume
	if err := os.MkdirAll(stagingPath, 0750); err != nil {
		klog.Error(log("blockMapper.stageVolumeForBlock failed to create dir %s: %v", stagingPath, err))
		return "", err
	}
	klog.V(4).Info(log("blockMapper.stageVolumeForBlock created stagingPath directory successfully [%s]", stagingPath))

	// Request to stage a block volume to stagingPath.
	// Expected implementation for driver is creating driver specific resource on stagingPath and
	// attaching the block volume to the node.
	err = csi.NodeStageVolume(ctx,
		csiSource.VolumeHandle,
		publishVolumeInfo,
		stagingPath,
		fsTypeBlockName,
		accessMode,
		nodeStageSecrets,
		csiSource.VolumeAttributes)

	if err != nil {
		klog.Error(log("blockMapper.stageVolumeForBlock failed: %v", err))
		return "", err
	}

	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock successfully requested NodeStageVolume [%s]", stagingPath))
	return stagingPath, nil
}
1199116
func (n *NodeInfo) VolumeLimits() map[v1.ResourceName]int64 {
	volumeLimits := map[v1.ResourceName]int64{}
	for k, v := range n.AllocatableResource().ScalarResources {
		if v1helper.IsAttachableVolumeResourceName(k) {
			volumeLimits[k] = v
		}
	}
	return volumeLimits
}
968298
func NewVolume(name, path string, pathType *v1.HostPathType) v1.Volume {
	return v1.Volume{
		Name: name,
		VolumeSource: v1.VolumeSource{
			HostPath: &v1.HostPathVolumeSource{
				Path: path,
				Type: pathType,
			},
		},
	}
}
1001159
func (s *VolumeService) GetPathForVolume(p *GetPathForVolumeParams) (*GetPathForVolumeResponse, error) {
	resp, err := s.cs.newRequest("getPathForVolume", p.toURLValues())
	if err != nil {
		return nil, err
	}

	var r GetPathForVolumeResponse
	if err := json.Unmarshal(resp, &r); err != nil {
		return nil, err
	}

	return &r, nil
}
727388
func (c *Client) UploadVolume(p *UploadVolumeParameter) (*Volume, error) {
	obj, err := c.Request("uploadVolume", convertParamToMap(p))
	if err != nil {
		return nil, err
	}
	return obj.(*Volume), err
}
1107626
func volumes(clientID C.h, attachments C.short) C.result {
	result := C.result{}

	c, err := getClient(clientID)
	if err != nil {
		result.err = C.CString(err.Error())
		return result
	}

	svcToVolMap, err := c.API().Volumes(
		nil, types.VolumeAttachmentsTypes(attachments))
	if err != nil {
		result.err = C.CString(err.Error())
		return result
	}

	service_names := []*C.char{}
	volume_maps := []*C.volume_map{}

	for service, volMap := range svcToVolMap {

		service_names = append(service_names, C.CString(service))

		volumes := []*C.volume{}
		volume_ids := []*C.char{}

		for volumeID, volume := range volMap {
			volume_ids = append(volume_ids, C.CString(volumeID))
			cVol, err := toCVolume(volume)
			if err != nil {
				result.err = C.CString(err.Error())
				return result
			}
			volumes = append(volumes, cVol)
		}

		lcVolMap := len(volMap)
		cVolMap := C.new_volume_map()
		cVolMap.volumes_c = C.int(lcVolMap)
		if lcVolMap > 0 {
			cVolMap.volume_ids = &volume_ids[0]
			cVolMap.volumes = &volumes[0]
		}
		volume_maps = append(volume_maps, cVolMap)
	}

	lcSvcVolMap := len(svcToVolMap)
	cSvcVolMap := C.new_service_volume_map()
	cSvcVolMap.services_c = C.int(lcSvcVolMap)
	if lcSvcVolMap > 0 {
		cSvcVolMap.service_names = &service_names[0]
		cSvcVolMap.volumes = &volume_maps[0]
	}

	result.val = unsafe.Pointer(cSvcVolMap)
	return result
}
131760
func (f *ControllerClient) ControllerUnpublishVolume(ctx context.Context, in *csipb.ControllerUnpublishVolumeRequest, opts ...grpc.CallOption) (*csipb.ControllerUnpublishVolumeResponse, error) {
	return nil, nil
}
1034262
func (s *OsdCsiServer) CreateVolume(
	ctx context.Context,
	req *csi.CreateVolumeRequest,
) (*csi.CreateVolumeResponse, error) {

	// Log request
	logrus.Debugf("CreateVolume req[%#v]", *req)

	if len(req.GetName()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "Name must be provided")
	}
	if req.GetVolumeCapabilities() == nil || len(req.GetVolumeCapabilities()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "Volume capabilities must be provided")
	}

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

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

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

	// Get Size
	if req.GetCapacityRange() != nil && req.GetCapacityRange().GetRequiredBytes() != 0 {
		spec.Size = uint64(req.GetCapacityRange().GetRequiredBytes())
	} else {
		spec.Size = defaultCSIVolumeSize
	}

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

	// Get secret if any was passed
	ctx = s.setupContextWithToken(ctx, req.GetSecrets())

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

	// Create volume
	var newVolumeId string
	if source.Parent == "" {
		// Get Capabilities and Size
		spec.Shared = csiRequestsSharedVolume(req)

		createResp, err := volumes.Create(ctx, &api.SdkVolumeCreateRequest{
			Name:   req.GetName(),
			Spec:   spec,
			Labels: locator.GetVolumeLabels(),
		})
		if err != nil {
			return nil, err
		}
		newVolumeId = createResp.VolumeId
	} else {
		cloneResp, err := volumes.Clone(ctx, &api.SdkVolumeCloneRequest{
			Name:     req.GetName(),
			ParentId: source.Parent,
		})
		if err != nil {
			return nil, err
		}
		newVolumeId = cloneResp.VolumeId
	}

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

	// Create response
	volume := &csi.Volume{}
	osdToCsiVolumeInfo(volume, inspectResp.GetVolume())
	return &csi.CreateVolumeResponse{
		Volume: volume,
	}, nil
}
864908
func (m *sioMgr) AttachVolume(volName string, multipleMappings bool) (string, error) {
	client, err := m.getClient()
	if err != nil {
		klog.Error(log("attach volume failed: %v", err))
		return "", err
	}

	klog.V(4).Infoln(log("attaching volume %s", volName))
	iid, err := client.IID()
	if err != nil {
		klog.Error(log("failed to get instanceID"))
		return "", err
	}
	klog.V(4).Info(log("attaching volume %s to host instance %s", volName, iid))

	devs, err := client.Devs()
	if err != nil {
		return "", err
	}

	vol, err := client.FindVolume(volName)
	if err != nil {
		klog.Error(log("failed to find volume %s: %v", volName, err))
		return "", err
	}

	// handle vol if already attached
	if len(vol.MappedSdcInfo) > 0 {
		if m.isSdcMappedToVol(iid, vol) {
			klog.V(4).Info(log("skippping attachment, volume %s already attached to sdc %s", volName, iid))
			return devs[vol.ID], nil
		}
	}

	// attach volume, get deviceName
	if err := client.AttachVolume(sioVolumeID(vol.ID), multipleMappings); err != nil {
		klog.Error(log("attachment for volume %s failed :%v", volName, err))
		return "", err
	}
	device, err := client.WaitForAttachedDevice(vol.ID)
	if err != nil {
		klog.Error(log("failed while waiting for device to attach: %v", err))
		return "", err
	}
	klog.V(4).Info(log("volume %s attached successfully as %s to instance %s", volName, device, iid))
	return device, nil
}
671754
func (s *OsdCsiServer) ControllerUnpublishVolume(
	context.Context,
	*csi.ControllerUnpublishVolumeRequest,
) (*csi.ControllerUnpublishVolumeResponse, error) {
	return nil, status.Error(codes.Unimplemented, "This request is not supported")
}
474649
func (os *OpenStack) DeleteVolume(volumeID string) error {
	used, err := os.diskIsUsed(volumeID)
	if err != nil {
		return err
	}
	if used {
		msg := fmt.Sprintf("Cannot delete the volume %q, it's still attached to a node", volumeID)
		return k8sVolume.NewDeletedVolumeInUseError(msg)
	}

	volumes, err := os.volumeService("")
	if err != nil || volumes == nil {
		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
		return err
	}

	err = volumes.deleteVolume(volumeID)
	if err != nil {
		glog.Errorf("Cannot delete volume %s: %v", volumeID, err)
	}
	return nil

}
1247198
func (f *NodeClient) NodeUnstageVolume(ctx context.Context, req *csipb.NodeUnstageVolumeRequest, opts ...grpc.CallOption) (*csipb.NodeUnstageVolumeResponse, error) {
	if f.nextErr != nil {
		return nil, f.nextErr
	}

	if req.GetVolumeId() == "" {
		return nil, errors.New("missing volume id")
	}
	if req.GetStagingTargetPath() == "" {
		return nil, errors.New("missing staging target path")
	}

	delete(f.nodeStagedVolumes, req.GetVolumeId())
	return &csipb.NodeUnstageVolumeResponse{}, nil
}
1098305
func (u *storageosUtil) MountVolume(b *storageosMounter, mntDevice, deviceMountPath string) error {
	notMnt, err := b.mounter.IsLikelyNotMountPoint(deviceMountPath)
	if err != nil {
		if os.IsNotExist(err) {
			if err = os.MkdirAll(deviceMountPath, 0750); err != nil {
				return err
			}
			notMnt = true
		} else {
			return err
		}
	}
	if err = os.MkdirAll(deviceMountPath, 0750); err != nil {
		klog.Errorf("mkdir failed on disk %s (%v)", deviceMountPath, err)
		return err
	}
	options := []string{}
	if b.readOnly {
		options = append(options, "ro")
	}
	if notMnt {
		err = b.diskMounter.FormatAndMount(mntDevice, deviceMountPath, b.fsType, options)
		if err != nil {
			os.Remove(deviceMountPath)
			return err
		}
	}
	return err
}
964365
func (s *OsdCsiServer) ControllerPublishVolume(
	context.Context,
	*csi.ControllerPublishVolumeRequest,
) (*csi.ControllerPublishVolumeResponse, error) {
	return nil, status.Error(codes.Unimplemented, "This request is not supported")
}
638384
func (c *sioClient) AttachVolume(id sioVolumeID, multipleMappings bool) error {
	if err := c.init(); err != nil {
		klog.Error(log("failed to init'd client in attach volume: %v", err))
		return err
	}

	iid, err := c.IID()
	if err != nil {
		klog.Error(log("failed to get instanceIID for attach volume: %v", err))
		return err
	}

	params := &siotypes.MapVolumeSdcParam{
		SdcID:                 iid,
		AllowMultipleMappings: strconv.FormatBool(multipleMappings),
		AllSdcs:               "",
	}
	volClient := sio.NewVolume(c.client)
	volClient.Volume = &siotypes.Volume{ID: string(id)}

	if err := volClient.MapVolumeSdc(params); err != nil {
		klog.Error(log("failed to attach volume id %s: %v", id, err))
		return err
	}

	klog.V(4).Info(log("volume %s attached successfully", id))
	return nil
}
56552
func (f *NodeClient) NodeGetVolumeStats(ctx context.Context, in *csipb.NodeGetVolumeStatsRequest, opts ...grpc.CallOption) (*csipb.NodeGetVolumeStatsResponse, error) {
	return nil, nil
}