// NodePublishVolume is a CSI API call which mounts the volume on the specified
// target path on the node.
//
// TODO: Support READ ONLY Mounts
//
766837
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 volume information
	v, err := util.VolumeFromName(s.driver, req.GetVolumeId())
	if err != nil {
		return nil, status.Errorf(codes.NotFound, "Volume id %s not found: %s",
			req.GetVolumeId(),
			err.Error())
	}
	if s.driver.Type() != 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.GetVolumeAttributes())
	if err != nil {
		return nil, status.Errorf(
			codes.InvalidArgument,
			"Invalid volume attributes: %#v",
			req.GetVolumeAttributes())
	}

	// This seems weird as a way to change opts to map[string]string
	opts := make(map[string]string)
	if len(spec.GetPassphrase()) != 0 {
		opts[options.OptionsSecret] = spec.GetPassphrase()
	}

	// If this is for a block driver, first attach the volume
	var devicePath string
	if s.driver.Type() == api.DriverType_DRIVER_TYPE_BLOCK {
		if devicePath, err = s.driver.Attach(req.GetVolumeId(), opts); err != nil {
			return nil, status.Errorf(
				codes.Internal,
				"Unable to attach volume: %s",
				err.Error())
		}
	}

	if req.GetVolumeCapability().GetBlock() != nil {
		// As block create a sym link to the attached location
		err = os.Symlink(devicePath, req.GetTargetPath())
		if err != nil {
			detachErr := s.driver.Detach(v.GetId(), opts)
			if detachErr != nil {
				logrus.Errorf("Unable to detach volume %s: %s",
					v.GetId(),
					detachErr.Error())
			}
			return nil, status.Errorf(
				codes.Internal,
				"Failed to create symlink %s -> %s: %v",
				req.GetTargetPath(),
				devicePath,
				err)
		}
	} else {
		// 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 := s.driver.Mount(req.GetVolumeId(), req.GetTargetPath(), nil); err != nil {
			// Detach on error
			detachErr := s.driver.Detach(v.GetId(), opts)
			if detachErr != nil {
				logrus.Errorf("Unable to detach volume %s: %s",
					v.GetId(),
					detachErr.Error())
			}
			return nil, status.Errorf(
				codes.Internal,
				"Unable to mount volume %s onto %s: %s",
				req.GetVolumeId(),
				req.GetTargetPath(),
				err.Error())
		}
	}

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

	return &csi.NodePublishVolumeResponse{}, nil
}
710580
func (s *VolumeService) CreateVolume(p *CreateVolumeParams) (*CreateVolumeResponse, error) {
	resp, err := s.cs.newRequest("createVolume", p.toURLValues())
	if err != nil {
		return nil, err
	}

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

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		b, err = getRawValue(b)
		if err != nil {
			return nil, err
		}

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

	return &r, nil
}
409280
func (c *client) MountBlob(ctx context.Context, sourceRef reference.Canonical, targetRef reference.Named) error {
	repoEndpoint, err := newDefaultRepositoryEndpoint(targetRef, c.insecureRegistry)
	if err != nil {
		return err
	}
	repo, err := c.getRepositoryForReference(ctx, targetRef, repoEndpoint)
	if err != nil {
		return err
	}
	lu, err := repo.Blobs(ctx).Create(ctx, distributionclient.WithMountFrom(sourceRef))
	switch err.(type) {
	case distribution.ErrBlobMounted:
		logrus.Debugf("mount of blob %s succeeded", sourceRef)
		return nil
	case nil:
	default:
		return errors.Wrapf(err, "failed to mount blob %s to %s", sourceRef, targetRef)
	}
	lu.Cancel(ctx)
	logrus.Debugf("mount of blob %s created", sourceRef)
	return ErrBlobCreated{From: sourceRef, Target: targetRef}
}
1238718
func (s *VolumeService) AttachVolume(p *AttachVolumeParams) (*AttachVolumeResponse, error) {
	resp, err := s.cs.newRequest("attachVolume", p.toURLValues())
	if err != nil {
		return nil, err
	}

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

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		b, err = getRawValue(b)
		if err != nil {
			return nil, err
		}

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

	return &r, nil
}
620901
func FindVolume(key string) *Volume {
	collections.volumeMutex.Lock()
	defer collections.volumeMutex.Unlock()

	obj := collections.volumes[key]
	if obj == nil {
		return nil
	}

	return obj
}
461757
func (s *VolumeService) ResizeVolume(p *ResizeVolumeParams) (*ResizeVolumeResponse, error) {
	resp, err := s.cs.newRequest("resizeVolume", p.toURLValues())
	if err != nil {
		return nil, err
	}

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

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		b, err = getRawValue(b)
		if err != nil {
			return nil, err
		}

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

	return &r, nil
}
1072922
func MultiPublish(topic string, bodies [][]byte) (*Command, error) {
	var params = [][]byte{[]byte(topic)}

	num := uint32(len(bodies))
	bodySize := 4
	for _, b := range bodies {
		bodySize += len(b) + 4
	}
	body := make([]byte, 0, bodySize)
	buf := bytes.NewBuffer(body)

	err := binary.Write(buf, binary.BigEndian, &num)
	if err != nil {
		return nil, err
	}
	for _, b := range bodies {
		err = binary.Write(buf, binary.BigEndian, int32(len(b)))
		if err != nil {
			return nil, err
		}
		_, err = buf.Write(b)
		if err != nil {
			return nil, err
		}
	}

	return &Command{[]byte("MPUB"), params, buf.Bytes()}, nil
}
101674
func (m *Manager) NewVolume(info *volume.Info) (volume.Volume, error) {
	return m.NewVolumeFromProvider("default", info)
}
808113
func (v *volumeSource) ReleaseVolumes(ctx context.ProviderCallContext, volIds []string) ([]error, error) {
	// noop
	return make([]error, len(volIds)), nil
}
1040080
func (v *OpenstackVolumes) AttachVolume(volume *Volume) error {
	opts := volumeattach.CreateOpts{
		VolumeID: volume.ID,
	}
	attachment, err := v.cloud.AttachVolume(v.meta.ServerID, opts)
	if err != nil {
		return fmt.Errorf("AttachVolume: failed to attach volume: %s", err)
	}
	volume.LocalDevice = attachment.Device
	return nil
}
530348
func mountVolume(pod *corev1.Pod, container *corev1.Container, objName, mountPath, volumeSuffix string, fsType policy.FSType) {
	volumeName := apihelpers.GetName(objName, volumeSuffix, kvalidation.DNS1123LabelMaxLength)

	// coerce from RFC1123 subdomain to RFC1123 label.
	volumeName = strings.Replace(volumeName, ".", "-", -1)

	volumeExists := false
	for _, v := range pod.Spec.Volumes {
		if v.Name == volumeName {
			volumeExists = true
			break
		}
	}
	mode := int32(0600)
	if !volumeExists {
		volume := makeVolume(volumeName, objName, mode, fsType)
		pod.Spec.Volumes = append(pod.Spec.Volumes, volume)
	}

	volumeMount := corev1.VolumeMount{
		Name:      volumeName,
		MountPath: mountPath,
		ReadOnly:  true,
	}
	container.VolumeMounts = append(container.VolumeMounts, volumeMount)
}
689572
func (c *Client) processPublish(publish *packet.Publish) error {
	// call callback for unacknowledged and directly acknowledged messages
	if publish.Message.QOS <= 1 {
		if c.Callback != nil {
			err := c.Callback(&publish.Message, nil)
			if err != nil {
				return c.die(err, true, true)
			}
		}
	}

	// handle qos 1 flow
	if publish.Message.QOS == 1 {
		// prepare puback packet
		puback := packet.NewPuback()
		puback.ID = publish.ID

		// acknowledge qos 1 publish
		err := c.send(puback, true)
		if err != nil {
			return c.die(err, false, false)
		}
	}

	// handle qos 2 flow
	if publish.Message.QOS == 2 {
		// store packet
		err := c.Session.SavePacket(session.Incoming, publish)
		if err != nil {
			return c.die(err, true, false)
		}

		// prepare pubrec packet
		pubrec := packet.NewPubrec()
		pubrec.ID = publish.ID

		// acknowledge qos 2 publish
		err = c.send(pubrec, true)
		if err != nil {
			return c.die(err, false, false)
		}
	}

	return nil
}
813640
func (c *Client) PutVolume(vol *ct.Volume) error {
	if vol.ID == "" {
		return errors.New("controller: missing id")
	}
	return c.Put(fmt.Sprintf("/volumes/%s", vol.ID), vol, vol)
}
779046
func (c *Host) GetVolume(volumeID string) (*volume.Info, error) {
	var volume volume.Info
	return &volume, c.c.Get(fmt.Sprintf("/storage/volumes/%s", volumeID), &volume)
}
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
}
1298631
func (f *NodeClient) NodeStageVolume(ctx context.Context, req *csipb.NodeStageVolumeRequest, opts ...grpc.CallOption) (*csipb.NodeStageVolumeResponse, 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")
	}

	fsType := ""
	fsTypes := "block|ext4|xfs|zfs"
	mounted := req.GetVolumeCapability().GetMount()
	if mounted != nil {
		fsType = mounted.GetFsType()
	}
	if !strings.Contains(fsTypes, fsType) {
		return nil, errors.New("invalid fstype")
	}

	f.nodeStagedVolumes[req.GetVolumeId()] = CSIVolume{
		Path:          req.GetStagingTargetPath(),
		VolumeContext: req.GetVolumeContext(),
	}
	return &csipb.NodeStageVolumeResponse{}, nil
}
510573
func (s *OsdCsiServer) NodeUnpublishVolume(
	ctx context.Context,
	req *csi.NodeUnpublishVolumeRequest,
) (*csi.NodeUnpublishVolumeResponse, error) {

	logrus.Debugf("NodeUnPublishVolume 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")
	}

	// Get volume information
	_, err := util.VolumeFromName(s.driver, req.GetVolumeId())
	if err != nil {
		return nil, status.Errorf(codes.NotFound, "Volume id %s not found: %s",
			req.GetVolumeId(),
			err.Error())
	}

	// Get information about the target since the request does not
	// tell us if it is for block or mount point.
	// https://github.com/container-storage-interface/spec/issues/285
	fileInfo, err := os.Lstat(req.GetTargetPath())
	if err != nil && os.IsNotExist(err) {
		// For idempotency, return that there is nothing to unmount
		logrus.Infof("NodeUnpublishVolume on target path %s but it does "+
			"not exist, returning there is nothing to do", req.GetTargetPath())
		return &csi.NodeUnpublishVolumeResponse{}, nil
	} else if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"Unknown error while verifying target location %s: %s",
			req.GetTargetPath(),
			err.Error())
	}

	// Check if it is block or not
	if fileInfo.Mode()&os.ModeSymlink != 0 {
		// If block, we just need to remove the link.
		os.Remove(req.GetTargetPath())
	} else {
		if !fileInfo.IsDir() {
			return nil, status.Errorf(
				codes.NotFound,
				"Target location %s is not a directory", req.GetTargetPath())
		}

		// Mount volume onto the path
		if err = s.driver.Unmount(req.GetVolumeId(), req.GetTargetPath(), nil); err != nil {
			return nil, status.Errorf(
				codes.Internal,
				"Unable to unmount volume %s onto %s: %s",
				req.GetVolumeId(),
				req.GetTargetPath(),
				err.Error())
		}
	}

	if s.driver.Type() == api.DriverType_DRIVER_TYPE_BLOCK {
		if err = s.driver.Detach(req.GetVolumeId(), nil); err != nil {
			return nil, status.Errorf(
				codes.Internal,
				"Unable to detach volume: %s",
				err.Error())
		}
	}

	logrus.Infof("Volume %s unmounted", req.GetVolumeId())

	return &csi.NodeUnpublishVolumeResponse{}, nil
}