// ControllerPublishVolume implements csi method
17050
func (f *ControllerClient) ControllerPublishVolume(ctx context.Context, in *csipb.ControllerPublishVolumeRequest, opts ...grpc.CallOption) (*csipb.ControllerPublishVolumeResponse, error) {
	return nil, nil
}
1134268
func httpCreateVolume(w http.ResponseWriter, r *http.Request, vars map[string]string) (interface{}, error) {
	log.Debugf("Received httpGetVolume: %+v", vars)

	var obj Volume
	key := vars["key"]

	// Get object from the request
	err := json.NewDecoder(r.Body).Decode(&obj)
	if err != nil {
		log.Errorf("Error decoding volume create request. Err %v", err)
		return nil, err
	}

	// set the key
	obj.Key = key

	// Create the object
	err = CreateVolume(&obj)
	if err != nil {
		log.Errorf("CreateVolume error for: %+v. Err: %v", obj, err)
		return nil, err
	}

	// Return the obj
	return obj, nil
}
339531
func (s *OsdCsiServer) ControllerGetCapabilities(
	ctx context.Context,
	req *csi.ControllerGetCapabilitiesRequest,
) (*csi.ControllerGetCapabilitiesResponse, error) {

	// Creating and deleting volumes supported
	capCreateDeleteVolume := &csi.ControllerServiceCapability{
		Type: &csi.ControllerServiceCapability_Rpc{
			Rpc: &csi.ControllerServiceCapability_RPC{
				Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
			},
		},
	}

	// Creating and deleting snapshots
	capCreateDeleteSnapshot := &csi.ControllerServiceCapability{
		Type: &csi.ControllerServiceCapability_Rpc{
			Rpc: &csi.ControllerServiceCapability_RPC{
				Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
			},
		},
	}

	// ListVolumes supported
	capListVolumes := &csi.ControllerServiceCapability{
		Type: &csi.ControllerServiceCapability_Rpc{
			Rpc: &csi.ControllerServiceCapability_RPC{
				Type: csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
			},
		},
	}

	return &csi.ControllerGetCapabilitiesResponse{
		Capabilities: []*csi.ControllerServiceCapability{
			capCreateDeleteVolume,
			capCreateDeleteSnapshot,
			capListVolumes,
		},
	}, 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
}
652734
func NewController(p ControllerParameters) (*PersistentVolumeController, error) {
	eventRecorder := p.EventRecorder
	if eventRecorder == nil {
		broadcaster := record.NewBroadcaster()
		broadcaster.StartLogging(klog.Infof)
		broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: p.KubeClient.CoreV1().Events("")})
		eventRecorder = broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "persistentvolume-controller"})
	}

	controller := &PersistentVolumeController{
		volumes:                       newPersistentVolumeOrderedIndex(),
		claims:                        cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
		kubeClient:                    p.KubeClient,
		eventRecorder:                 eventRecorder,
		runningOperations:             goroutinemap.NewGoRoutineMap(true /* exponentialBackOffOnError */),
		cloud:                         p.Cloud,
		enableDynamicProvisioning:     p.EnableDynamicProvisioning,
		clusterName:                   p.ClusterName,
		createProvisionedPVRetryCount: createProvisionedPVRetryCount,
		createProvisionedPVInterval:   createProvisionedPVInterval,
		claimQueue:                    workqueue.NewNamed("claims"),
		volumeQueue:                   workqueue.NewNamed("volumes"),
		resyncPeriod:                  p.SyncPeriod,
	}

	// Prober is nil because PV is not aware of Flexvolume.
	if err := controller.volumePluginMgr.InitPlugins(p.VolumePlugins, nil /* prober */, controller); err != nil {
		return nil, fmt.Errorf("Could not initialize volume plugins for PersistentVolume Controller: %v", err)
	}

	p.VolumeInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj interface{}) { controller.enqueueWork(controller.volumeQueue, obj) },
			UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(controller.volumeQueue, newObj) },
			DeleteFunc: func(obj interface{}) { controller.enqueueWork(controller.volumeQueue, obj) },
		},
	)
	controller.volumeLister = p.VolumeInformer.Lister()
	controller.volumeListerSynced = p.VolumeInformer.Informer().HasSynced

	p.ClaimInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj interface{}) { controller.enqueueWork(controller.claimQueue, obj) },
			UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(controller.claimQueue, newObj) },
			DeleteFunc: func(obj interface{}) { controller.enqueueWork(controller.claimQueue, obj) },
		},
	)
	controller.claimLister = p.ClaimInformer.Lister()
	controller.claimListerSynced = p.ClaimInformer.Informer().HasSynced

	controller.classLister = p.ClassInformer.Lister()
	controller.classListerSynced = p.ClassInformer.Informer().HasSynced
	controller.podLister = p.PodInformer.Lister()
	controller.podListerSynced = p.PodInformer.Informer().HasSynced
	controller.NodeLister = p.NodeInformer.Lister()
	controller.NodeListerSynced = p.NodeInformer.Informer().HasSynced
	return controller, 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
}
1090832
func controllerCinderVolumes(storageAdapter OpenstackStorage, controllerUUID string) ([]cinder.Volume, error) {
	return cinderVolumes(storageAdapter, func(v *cinder.Volume) bool {
		return v.Metadata[tags.JujuController] == controllerUUID
	})
}
120590
func chPublish(c *EncodedConn, chVal reflect.Value, subject string) {
	for {
		val, ok := chVal.Recv()
		if !ok {
			// Channel has most likely been closed.
			return
		}
		if e := c.Publish(subject, val.Interface()); e != nil {
			// Do this under lock.
			c.Conn.mu.Lock()
			defer c.Conn.mu.Unlock()

			if c.Conn.Opts.AsyncErrorCB != nil {
				// FIXME(dlc) - Not sure this is the right thing to do.
				// FIXME(ivan) - If the connection is not yet closed, try to schedule the callback
				if c.Conn.isClosed() {
					go c.Conn.Opts.AsyncErrorCB(c.Conn, nil, e)
				} else {
					c.Conn.ach.push(func() { c.Conn.Opts.AsyncErrorCB(c.Conn, nil, e) })
				}
			}
			return
		}
	}
}
354761
func (c *Client) NewVolume(opts *NewVolumeOptions) (NewVolumeResult, error) {
	req := xms.PostVolumesReq(*opts)
	return c.api.PostVolumes(&req)
}
105487
func (util *portworxVolumeUtil) AttachVolume(m *portworxVolumeMounter, attachOptions map[string]string) (string, error) {
	driver, err := util.getLocalPortworxDriver(m.plugin.host)
	if err != nil || driver == nil {
		klog.Errorf("Failed to get portworx driver. Err: %v", err)
		return "", err
	}

	devicePath, err := driver.Attach(m.volName, attachOptions)
	if err != nil {
		klog.Errorf("Error attaching Portworx Volume (%v): %v", m.volName, err)
		return "", err
	}
	return devicePath, nil
}
671754
func (s *OsdCsiServer) ControllerUnpublishVolume(
	context.Context,
	*csi.ControllerUnpublishVolumeRequest,
) (*csi.ControllerUnpublishVolumeResponse, error) {
	return nil, status.Error(codes.Unimplemented, "This request is not supported")
}
956054
func (util *portworxVolumeUtil) MountVolume(m *portworxVolumeMounter, mountPath string) error {
	driver, err := util.getLocalPortworxDriver(m.plugin.host)
	if err != nil || driver == nil {
		klog.Errorf("Failed to get portworx driver. Err: %v", err)
		return err
	}

	err = driver.Mount(m.volName, mountPath)
	if err != nil {
		klog.Errorf("Error mounting Portworx Volume (%v) on Path (%v): %v", m.volName, mountPath, err)
		return err
	}
	return nil
}
483117
func getVolumeDevicePluginDir(specVolID string, host volume.VolumeHost) string {
	sanitizedSpecVolID := utilstrings.EscapeQualifiedName(specVolID)
	return filepath.Join(host.GetVolumeDevicePluginDir(CSIPluginName), sanitizedSpecVolID, "dev")
}
1089721
func MountPublicController(service *goa.Service, ctrl PublicController) {
	initService(service)
	var h goa.Handler
	service.Mux.Handle("OPTIONS", "/ui", ctrl.MuxHandler("preflight", handlePublicOrigin(cors.HandlePreflight()), nil))

	h = ctrl.FileHandler("/ui", "public/html/index.html")
	h = handlePublicOrigin(h)
	service.Mux.Handle("GET", "/ui", ctrl.MuxHandler("serve", h, nil))
	service.LogInfo("mount", "ctrl", "Public", "files", "public/html/index.html", "route", "GET /ui")
}
711352
func (e *JSONEncoder) EncodePublish(res *PublishResult) ([]byte, error) {
	return json.Marshal(res)
}
413708
func CompressFilter(c *Controller, fc []Filter) {
	if c.Response.Out.internalHeader.Server != nil && Config.BoolDefault("results.compressed", false) {
		if c.Response.Status != http.StatusNoContent && c.Response.Status != http.StatusNotModified {
			if found, compressType, compressWriter := detectCompressionType(c.Request, c.Response); found {
				writer := CompressResponseWriter{
					ControllerResponse: c.Response,
					OriginalWriter:     c.Response.GetWriter(),
					compressWriter:     compressWriter,
					compressionType:    compressType,
					headersWritten:     false,
					closeNotify:        make(chan bool, 1),
					closed:             false,
				}
				// Swap out the header with our own
				writer.Header = NewBufferedServerHeader(c.Response.Out.internalHeader.Server)
				c.Response.Out.internalHeader.Server = writer.Header
				if w, ok := c.Response.GetWriter().(http.CloseNotifier); ok {
					writer.parentNotify = w.CloseNotify()
				}
				c.Response.SetWriter(&writer)
			}
		} else {
			compressLog.Debug("CompressFilter: Compression disabled for response ", "status", c.Response.Status)
		}
	}
	fc[0](c, fc[1:])
}
1001051
func NewPUBLISHFromBytes(fixedHeader FixedHeader, remaining []byte) (Packet, error) {
	// Validate the byte data.
	if err := validatePUBLISHBytes(fixedHeader, remaining); err != nil {
		return nil, err
	}

	// Get the first byte from the fixedHeader.
	b := fixedHeader[0]

	// Create a PUBLISH Packet.
	p := &PUBLISH{
		DUP:    b&0x08 == 0x08,
		QoS:    b & 0x06 >> 1,
		retain: b&0x01 == 0x01,
	}

	// Set the fixed header to the Packet.
	p.fixedHeader = fixedHeader

	// Extract the length of the Topic Name.
	lenTopicName, _ := decodeUint16(remaining[0:2])

	// Calculate the length of the variable header.
	var lenVariableHeader int

	if p.QoS == mqtt.QoS0 {
		lenVariableHeader = 2 + int(lenTopicName)
	} else {
		lenVariableHeader = 2 + int(lenTopicName) + 2
	}

	// Set the variable header to the Packet.
	p.variableHeader = remaining[:lenVariableHeader]

	// Set the payload to the Packet.
	p.payload = remaining[lenVariableHeader:]

	// Set the Topic Name to the Packet.
	p.TopicName = remaining[2 : 2+lenTopicName]

	// Extract the Packet Identifier.
	var packetID uint16

	if p.QoS != mqtt.QoS0 {
		packetID, _ = decodeUint16(remaining[2+lenTopicName : 2+lenTopicName+2])
	}

	// Set the Packet Identifier to the Packet.
	p.PacketID = packetID

	// Set the Application Message to the Packet.
	p.Message = p.payload

	// Return the Packet.
	return p, nil
}