// NodePublishVolume implements CSI NodePublishVolume
968119
func (f *NodeClient) NodePublishVolume(ctx context.Context, req *csipb.NodePublishVolumeRequest, opts ...grpc.CallOption) (*csipb.NodePublishVolumeResponse, error) {
	if f.nextErr != nil {
		return nil, f.nextErr
	}

	if req.GetVolumeId() == "" {
		return nil, errors.New("missing volume id")
	}
	if req.GetTargetPath() == "" {
		return nil, errors.New("missing target path")
	}
	fsTypes := "block|ext4|xfs|zfs"
	fsType := req.GetVolumeCapability().GetMount().GetFsType()
	if !strings.Contains(fsTypes, fsType) {
		return nil, errors.New("invalid fstype")
	}
	f.nodePublishedVolumes[req.GetVolumeId()] = CSIVolume{
		VolumeHandle:    req.GetVolumeId(),
		Path:            req.GetTargetPath(),
		DeviceMountPath: req.GetStagingTargetPath(),
		VolumeContext:   req.GetVolumeContext(),
		FSType:          req.GetVolumeCapability().GetMount().GetFsType(),
		MountFlags:      req.GetVolumeCapability().GetMount().MountFlags,
	}
	return &csipb.NodePublishVolumeResponse{}, nil
}
478730
func (name KubernetesVolumeID) MapToAWSVolumeID() (EBSVolumeID, error) {
	awsID, err := csimigration.KubernetesVolumeIDToEBSVolumeID(string(name))
	if err != nil {
		return "", err
	}
	return EBSVolumeID(awsID), nil
}
225188
func (api *PublicAdminAPI) NodeInfo() (*p2p.NodeInfo, error) {
	server := api.node.Server()
	if server == nil {
		return nil, ErrNodeStopped
	}
	return server.NodeInfo(), nil
}
379062
func (v *version) CSINodes() CSINodeInformer {
	return &cSINodeInformer{factory: v.factory, tweakListOptions: v.tweakListOptions}
}
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
}
711352
func (e *JSONEncoder) EncodePublish(res *PublishResult) ([]byte, error) {
	return json.Marshal(res)
}
12795
func Convert_core_CSIPersistentVolumeSource_To_v1_CSIPersistentVolumeSource(in *core.CSIPersistentVolumeSource, out *v1.CSIPersistentVolumeSource, s conversion.Scope) error {
	return autoConvert_core_CSIPersistentVolumeSource_To_v1_CSIPersistentVolumeSource(in, out, s)
}
582402
func (os *OpenStack) CreateVolume(name string, size int, vtype, availability, snapshotID string, tags *map[string]string) (string, string, error) {
	volumes, err := os.volumeService("")
	if err != nil || volumes == nil {
		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
		return "", "", err
	}

	opts := VolumeCreateOpts{
		Name:             name,
		Size:             size,
		VolumeType:       vtype,
		Availability:     availability,
		SourceSnapshotID: snapshotID,
	}
	if tags != nil {
		opts.Metadata = *tags
	}

	volumeID, volumeAZ, err := volumes.createVolume(opts)

	if err != nil {
		glog.Errorf("Failed to create a %d GB volume: %v", size, err)
		return "", "", err
	}

	glog.Infof("Created volume %v in Availability Zone: %v", volumeID, volumeAZ)
	return volumeID, volumeAZ, nil
}
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
}
841449
func MakePublishInfo(networks map[string]string, ports []string) map[string][]string {
	result := map[string][]string{}
	for networkName, ip := range networks {
		data := []string{}
		for _, port := range ports {
			data = append(data, fmt.Sprintf("%s:%s", ip, port))
		}
		if len(data) > 0 {
			result[networkName] = data
		}
	}
	return result
}
462458
func (util *DiskUtil) CreateVolume(c *cinderVolumeProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (volumeID string, volumeSizeGB int, volumeLabels map[string]string, fstype string, err error) {
	cloud, err := c.plugin.getCloudProvider()
	if err != nil {
		return "", 0, nil, "", err
	}

	capacity := c.options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
	// Cinder works with gigabytes, convert to GiB with rounding up
	volSizeGiB, err := volumehelpers.RoundUpToGiBInt(capacity)
	if err != nil {
		return "", 0, nil, "", err
	}

	name := volutil.GenerateVolumeName(c.options.ClusterName, c.options.PVName, 255) // Cinder volume name can have up to 255 characters
	vtype := ""
	availability := ""
	// Apply ProvisionerParameters (case-insensitive). We leave validation of
	// the values to the cloud provider.
	for k, v := range c.options.Parameters {
		switch strings.ToLower(k) {
		case "type":
			vtype = v
		case "availability":
			availability = v
		case volume.VolumeParameterFSType:
			fstype = v
		default:
			return "", 0, nil, "", fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
		}
	}
	// TODO: implement PVC.Selector parsing
	if c.options.PVC.Spec.Selector != nil {
		return "", 0, nil, "", fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on Cinder")
	}

	if availability == "" {
		// No zone specified, choose one randomly in the same region
		zones, err := getZonesFromNodes(c.plugin.host.GetKubeClient())
		if err != nil {
			klog.V(2).Infof("error getting zone information: %v", err)
			return "", 0, nil, "", err
		}
		// if we did not get any zones, lets leave it blank and gophercloud will
		// use zone "nova" as default
		if len(zones) > 0 {
			availability, err = volumehelpers.SelectZoneForVolume(false, false, "", nil, zones, node, allowedTopologies, c.options.PVC.Name)
			if err != nil {
				klog.V(2).Infof("error selecting zone for volume: %v", err)
				return "", 0, nil, "", err
			}
		}
	}

	volumeID, volumeAZ, volumeRegion, IgnoreVolumeAZ, err := cloud.CreateVolume(name, volSizeGiB, vtype, availability, c.options.CloudTags)
	if err != nil {
		klog.V(2).Infof("Error creating cinder volume: %v", err)
		return "", 0, nil, "", err
	}
	klog.V(2).Infof("Successfully created cinder volume %s", volumeID)

	// these are needed that pod is spawning to same AZ
	volumeLabels = make(map[string]string)
	if IgnoreVolumeAZ == false {
		if volumeAZ != "" {
			volumeLabels[v1.LabelZoneFailureDomain] = volumeAZ
		}
		if volumeRegion != "" {
			volumeLabels[v1.LabelZoneRegion] = volumeRegion
		}
	}
	return volumeID, volSizeGiB, volumeLabels, fstype, nil
}
1173937
func (c *Client) processPublish(publish *packet.Publish) error {
	// handle qos 0 flow
	if publish.Message.QOS == 0 {
		// publish message
		err := c.backend.Publish(c, &publish.Message, nil)
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)

		return nil
	}

	// acquire publish token
	select {
	case <-c.publishTokens:
		// continue
	case <-time.After(c.TokenTimeout):
		return c.die(ClientError, ErrTokenTimeout)
	case <-c.tomb.Dying():
		return tomb.ErrDying
	}

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

		// publish message and queue puback if ack is called
		err := c.backend.Publish(c, &publish.Message, func() {
			c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)

			select {
			case c.ackQueue <- puback:
			case <-c.tomb.Dying():
			}
		})
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
	}

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

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

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

	return nil
}
408827
func (c *Client) AttachVolume(p *AttachVolumeParameter) (*Volume, error) {
	obj, err := c.Request("attachVolume", convertParamToMap(p))
	if err != nil {
		return nil, err
	}
	return obj.(*Volume), err
}
671754
func (s *OsdCsiServer) ControllerUnpublishVolume(
	context.Context,
	*csi.ControllerUnpublishVolumeRequest,
) (*csi.ControllerUnpublishVolumeResponse, error) {
	return nil, status.Error(codes.Unimplemented, "This request is not supported")
}
316896
func modelCinderVolumes(storageAdapter OpenstackStorage, modelUUID string) ([]cinder.Volume, error) {
	return cinderVolumes(storageAdapter, func(v *cinder.Volume) bool {
		return v.Metadata[tags.JujuModel] == modelUUID
	})
}
611563
func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
	var nodeIsPublic bool
	err := c.db.View(func(tx *bbolt.Tx) error {
		nodes := tx.Bucket(nodeBucket)
		if nodes == nil {
			return ErrGraphNodesNotFound
		}
		ourPubKey := nodes.Get(sourceKey)
		if ourPubKey == nil {
			return ErrSourceNodeNotSet
		}
		node, err := fetchLightningNode(nodes, pubKey[:])
		if err != nil {
			return err
		}

		nodeIsPublic, err = node.isPublic(tx, ourPubKey)
		return err
	})
	if err != nil {
		return false, err
	}

	return nodeIsPublic, nil
}
546309
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 Size
	if req.GetCapacityRange() != nil && req.GetCapacityRange().GetRequiredBytes() != 0 {
		spec.Size = uint64(req.GetCapacityRange().GetRequiredBytes())
	} else {
		spec.Size = defaultCSIVolumeSize
	}

	// Create response
	volume := &csi.Volume{}
	resp := &csi.CreateVolumeResponse{
		Volume: volume,
	}

	// Check if the volume has already been created or is in process of creation
	v, err := util.VolumeFromName(s.driver, req.GetName())
	if err == nil {
		// Check the requested arguments match that of the existing volume
		if spec.Size != v.GetSpec().GetSize() {
			return nil, status.Errorf(
				codes.AlreadyExists,
				"Existing volume has a size of %v which differs from requested size of %v",
				v.GetSpec().GetSize(),
				spec.Size)
		}
		if v.GetSpec().GetShared() != csiRequestsSharedVolume(req) {
			return nil, status.Errorf(
				codes.AlreadyExists,
				"Existing volume has shared=%v while request is asking for shared=%v",
				v.GetSpec().GetShared(),
				csiRequestsSharedVolume(req))
		}
		if v.GetSource().GetParent() != source.GetParent() {
			return nil, status.Error(codes.AlreadyExists, "Existing volume has conflicting parent value")
		}

		// Return information on existing volume
		osdToCsiVolumeInfo(volume, v)
		return resp, nil
	}

	// Check if this is a cloning request to create a volume from a snapshot
	if req.GetVolumeContentSource().GetSnapshot() != nil {
		source.Parent = req.GetVolumeContentSource().GetSnapshot().GetId()
	}

	// Check if the caller is asking to create a snapshot or for a new volume
	var id string
	if source != nil && len(source.GetParent()) != 0 {
		// Get parent volume information
		parent, err := util.VolumeFromName(s.driver, source.Parent)
		if err != nil {
			e := fmt.Sprintf("unable to get parent volume information: %s\n", err.Error())
			logrus.Errorln(e)
			return nil, status.Error(codes.InvalidArgument, e)
		}

		// Create a snapshot from the parent
		id, err = s.driver.Snapshot(parent.GetId(), false, &api.VolumeLocator{
			Name: req.GetName(),
		},
			false)
		if err != nil {
			e := fmt.Sprintf("unable to create snapshot: %s\n", err.Error())
			logrus.Errorln(e)
			return nil, status.Error(codes.Internal, e)
		}
	} else {
		// Get Capabilities and Size
		spec.Shared = csiRequestsSharedVolume(req)

		// Create the volume
		locator.Name = req.GetName()

		// get enforced policy specs
		// 0.3 Does not support user context or auth
		spec, err = sdkVol.GetDefaultVolSpecs(context.Background(), spec, false /* not an update */)
		if err != nil {
			return nil, status.Errorf(codes.Internal, err.Error())
		}
		id, err = s.driver.Create(locator, source, spec)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
	}

	// id must have been set
	v, err = util.VolumeFromName(s.driver, id)
	if err != nil {
		e := fmt.Sprintf("Unable to find newly created volume: %s", err.Error())
		logrus.Errorln(e)
		return nil, status.Error(codes.Internal, e)
	}
	osdToCsiVolumeInfo(volume, v)
	return resp, nil
}