// DiscoverLocalVolumes reads the configured discovery paths, and creates PVs for the new volumes
19137
func (d *Discoverer) DiscoverLocalVolumes() {
	for class, config := range d.DiscoveryMap {
		d.discoverVolumesAtPath(class, config)
	}
}
31653
func (s *oracleVolumeSource) CreateVolumes(ctx context.ProviderCallContext, params []storage.VolumeParams) ([]storage.CreateVolumesResult, error) {
	if params == nil {
		return []storage.CreateVolumesResult{}, nil
	}
	results := make([]storage.CreateVolumesResult, len(params))
	for i, volume := range params {
		vol, err := s.createVolume(volume)
		if err != nil {
			results[i].Error = errors.Trace(err)
			continue
		}
		results[i].Volume = vol
	}
	return results, nil
}
759115
func (s *cinderVolumeSource) CreateVolumes(ctx context.ProviderCallContext, args []storage.VolumeParams) ([]storage.CreateVolumesResult, error) {
	results := make([]storage.CreateVolumesResult, len(args))
	for i, arg := range args {
		volume, err := s.createVolume(arg)
		if err != nil {
			results[i].Error = errors.Trace(err)
			if denied := common.MaybeHandleCredentialError(IsAuthorisationFailure, err, ctx); denied {
				// If it is an unauthorised error, no need to continue since we will 100% fail...
				break
			}
			continue
		}
		results[i].Volume = volume
	}
	return results, nil
}
1153561
func (v *ebsVolumeSource) CreateVolumes(ctx context.ProviderCallContext, params []storage.VolumeParams) (_ []storage.CreateVolumesResult, err error) {

	// First, validate the params before we use them.
	results := make([]storage.CreateVolumesResult, len(params))
	instanceIds := set.NewStrings()
	for i, p := range params {
		if err := v.ValidateVolumeParams(p); err != nil {
			results[i].Error = err
			continue
		}
		instanceIds.Add(string(p.Attachment.InstanceId))
	}

	instances := make(instanceCache)
	if instanceIds.Size() > 1 {
		if err := instances.update(v.env.ec2, ctx, instanceIds.Values()...); err != nil {
			err := maybeConvertCredentialError(err, ctx)
			logger.Debugf("querying running instances: %v", err)
			// We ignore the error, because we don't want an invalid
			// InstanceId reference from one VolumeParams to prevent
			// the creation of another volume.
			// Except if it is a credential error...
			if common.IsCredentialNotValid(err) {
				return nil, errors.Trace(err)
			}
		}
	}

	for i, p := range params {
		if results[i].Error != nil {
			continue
		}
		volume, attachment, err := v.createVolume(ctx, p, instances)
		if err != nil {
			results[i].Error = err
			continue
		}
		results[i].Volume = volume
		results[i].VolumeAttachment = attachment
	}
	return results, nil
}
103612
func (v *OpenstackVolumes) FindMountedVolume(volume *Volume) (string, error) {
	device := volume.LocalDevice

	_, err := os.Stat(pathFor(device))
	if err == nil {
		return device, nil
	}
	if os.IsNotExist(err) {
		return "", nil
	}
	return "", fmt.Errorf("error checking for device %q: %v", device, err)
}
1255355
func (s *oracleVolumeSource) AttachVolumes(ctx context.ProviderCallContext, params []storage.VolumeAttachmentParams) ([]storage.AttachVolumesResult, error) {
	instanceIds := []instance.Id{}
	for _, val := range params {
		instanceIds = append(instanceIds, val.InstanceId)
	}
	if len(instanceIds) == 0 {
		return []storage.AttachVolumesResult{}, nil
	}
	instancesAsMap, err := s.env.getOracleInstancesAsMap(instanceIds...)
	if err != nil {
		return []storage.AttachVolumesResult{}, errors.Trace(err)
	}
	attachmentsAsMap, err := s.getStorageAttachments()
	if err != nil {
		return []storage.AttachVolumesResult{}, errors.Trace(err)
	}

	ret := make([]storage.AttachVolumesResult, len(params))

	for i, val := range params {
		instance, ok := instancesAsMap[string(val.InstanceId)]
		if !ok {
			ret[i].Error = errors.NotFoundf("instance %q was not found", string(val.InstanceId))
			continue
		}

		result, err := s.attachVolume(instance, attachmentsAsMap, val)
		if err != nil {
			ret[i].Error = errors.Trace(err)
			continue
		}
		ret[i] = result

	}
	logger.Infof("returning attachments: %v", ret)
	return ret, nil
}
384542
func NewDiscoveryClientForConfig(c *Config) (*DiscoveryClient, error) {
	config := *c
	if err := setDiscoveryDefaults(&config); err != nil {
		return nil, err
	}
	client, err := UnversionedRESTClientFor(&config)
	return &DiscoveryClient{client}, err
}
245419
func NewOpenstackVolumes() (*OpenstackVolumes, error) {

	metadata, err := getLocalMetadata()
	if err != nil {
		return nil, fmt.Errorf("Failed to get server metadata: %v", err)
	}

	tags := make(map[string]string)
	// Cluster name needed to bypass missing designate options
	tags[openstack.TagClusterName] = metadata.UserMeta.ClusterName

	oscloud, err := openstack.NewOpenstackCloud(tags, nil)
	if err != nil {
		return nil, fmt.Errorf("Failed to initialize OpenstackVolumes: %v", err)
	}

	a := &OpenstackVolumes{
		cloud: oscloud,
		meta:  metadata,
	}

	err = a.discoverTags()
	if err != nil {
		return nil, err
	}

	return a, nil
}
881763
func GetVolumeAPIRoutes(name, sdkUds string) []*Route {
	volMgmtApi := newVolumeAPI(name, sdkUds)
	return volMgmtApi.Routes()
}
1171028
func Convert_v1beta2_Discovery_To_kubeadm_Discovery(in *Discovery, out *kubeadm.Discovery, s conversion.Scope) error {
	return autoConvert_v1beta2_Discovery_To_kubeadm_Discovery(in, out, s)
}
130336
func StartLocalController(client *kubernetes.Clientset, ptable deleter.ProcTable, config *common.UserConfig) {
	glog.Info("Initializing volume cache\n")

	var provisionerName string
	if config.UseNodeNameOnly {
		provisionerName = fmt.Sprintf("local-volume-provisioner-%v", config.Node.Name)
	} else {
		provisionerName = fmt.Sprintf("local-volume-provisioner-%v-%v", config.Node.Name, config.Node.UID)
	}

	broadcaster := record.NewBroadcaster()
	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(client.CoreV1().RESTClient()).Events("")})
	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: provisionerName})

	// We choose a random resync period between MinResyncPeriod and 2 *
	// MinResyncPeriod, so that local provisioners deployed on multiple nodes
	// at same time don't list the apiserver simultaneously.
	resyncPeriod := time.Duration(config.MinResyncPeriod.Seconds()*(1+rand.Float64())) * time.Second

	runtimeConfig := &common.RuntimeConfig{
		UserConfig:      config,
		Cache:           cache.NewVolumeCache(),
		VolUtil:         util.NewVolumeUtil(),
		APIUtil:         util.NewAPIUtil(client),
		Client:          client,
		Name:            provisionerName,
		Recorder:        recorder,
		Mounter:         mount.New("" /* default mount path */),
		InformerFactory: informers.NewSharedInformerFactory(client, resyncPeriod),
	}

	populator.NewPopulator(runtimeConfig)

	var jobController deleter.JobController
	var err error
	if runtimeConfig.UseJobForCleaning {
		labels := map[string]string{common.NodeNameLabel: config.Node.Name}
		jobController, err = deleter.NewJobController(labels, runtimeConfig)
		if err != nil {
			glog.Fatalf("Error initializing jobController: %v", err)
		}
		glog.Infof("Enabling Jobs based cleaning.")
	}
	cleanupTracker := &deleter.CleanupStatusTracker{ProcTable: ptable, JobController: jobController}

	discoverer, err := discovery.NewDiscoverer(runtimeConfig, cleanupTracker)
	if err != nil {
		glog.Fatalf("Error initializing discoverer: %v", err)
	}

	deleter := deleter.NewDeleter(runtimeConfig, cleanupTracker)

	// Start informers after all event listeners are registered.
	runtimeConfig.InformerFactory.Start(wait.NeverStop)
	// Wait for all started informers' cache were synced.
	for v, synced := range runtimeConfig.InformerFactory.WaitForCacheSync(wait.NeverStop) {
		if !synced {
			glog.Fatalf("Error syncing informer for %v", v)
		}
	}
	// Run controller logic.
	if jobController != nil {
		go jobController.Run(wait.NeverStop)
	}
	glog.Info("Controller started\n")
	for {
		deleter.DeletePVs()
		discoverer.DiscoverLocalVolumes()
		time.Sleep(10 * time.Second)
	}
}
1343523
func (dp *MockStaticDiscoveryProvider) CreateLocalDiscoveryService(mspID string) (fab.DiscoveryService, error) {

	if dp.customDiscoveryService != nil {
		return dp.customDiscoveryService, nil
	}

	return NewMockDiscoveryService(dp.Error, dp.Peers...), nil
}
205391
func (v *version) PersistentVolumes() PersistentVolumeInformer {
	return &persistentVolumeInformer{factory: v.factory, tweakListOptions: v.tweakListOptions}
}
1129167
func createVolumes(ctx *context, ops map[names.VolumeTag]*createVolumeOp) error {
	volumeParams := make([]storage.VolumeParams, 0, len(ops))
	for _, op := range ops {
		volumeParams = append(volumeParams, op.args)
	}
	paramsBySource, volumeSources, err := volumeParamsBySource(
		ctx.config.StorageDir, volumeParams, ctx.config.Registry,
	)
	if err != nil {
		return errors.Trace(err)
	}
	var reschedule []scheduleOp
	var volumes []storage.Volume
	var volumeAttachments []storage.VolumeAttachment
	var statuses []params.EntityStatusArgs
	for sourceName, volumeParams := range paramsBySource {
		logger.Debugf("creating volumes: %v", volumeParams)
		volumeSource := volumeSources[sourceName]
		validVolumeParams, validationErrors := validateVolumeParams(volumeSource, volumeParams)
		for i, err := range validationErrors {
			if err == nil {
				continue
			}
			statuses = append(statuses, params.EntityStatusArgs{
				Tag:    volumeParams[i].Tag.String(),
				Status: status.Error.String(),
				Info:   err.Error(),
			})
			logger.Debugf(
				"failed to validate parameters for %s: %v",
				names.ReadableString(volumeParams[i].Tag), err,
			)
		}
		volumeParams = validVolumeParams
		if len(volumeParams) == 0 {
			continue
		}
		results, err := volumeSource.CreateVolumes(ctx.config.CloudCallContext, volumeParams)
		if err != nil {
			return errors.Annotatef(err, "creating volumes from source %q", sourceName)
		}
		for i, result := range results {
			statuses = append(statuses, params.EntityStatusArgs{
				Tag:    volumeParams[i].Tag.String(),
				Status: status.Attaching.String(),
			})
			entityStatus := &statuses[len(statuses)-1]
			if result.Error != nil {
				// Reschedule the volume creation.
				reschedule = append(reschedule, ops[volumeParams[i].Tag])

				// Note: we keep the status as "pending" to indicate
				// that we will retry. When we distinguish between
				// transient and permanent errors, we will set the
				// status to "error" for permanent errors.
				entityStatus.Status = status.Pending.String()
				entityStatus.Info = result.Error.Error()
				logger.Debugf(
					"failed to create %s: %v",
					names.ReadableString(volumeParams[i].Tag),
					result.Error,
				)
				continue
			}
			volumes = append(volumes, *result.Volume)
			if result.VolumeAttachment != nil {
				entityStatus.Status = status.Attached.String()
				volumeAttachments = append(volumeAttachments, *result.VolumeAttachment)
			}
		}
	}
	scheduleOperations(ctx, reschedule...)
	setStatus(ctx, statuses)
	if len(volumes) == 0 {
		return nil
	}
	// TODO(axw) we need to be able to list volumes in the provider,
	// by environment, so that we can "harvest" them if they're
	// unknown. This will take care of killing volumes that we fail
	// to record in state.
	errorResults, err := ctx.config.Volumes.SetVolumeInfo(volumesFromStorage(volumes))
	if err != nil {
		return errors.Annotate(err, "publishing volumes to state")
	}
	for i, result := range errorResults {
		if result.Error != nil {
			logger.Errorf(
				"publishing volume %s to state: %v",
				volumes[i].Tag.Id(),
				result.Error,
			)
		}
	}
	for _, v := range volumes {
		updateVolume(ctx, v)
	}
	// Note: the storage provisioner that creates a volume is also
	// responsible for creating the volume attachment. It is therefore
	// safe to set the volume attachment info after the volume info,
	// without leading to the possibility of concurrent, duplicate
	// attachments.
	err = setVolumeAttachmentInfo(ctx, volumeAttachments)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}
849520
func WithLocalPeers(peers ...*MockDiscoveryPeerEndpoint) MockDiscoveryServerOpt {
	return func(s *MockDiscoveryServer) {
		s.localPeersByOrg = asPeersByOrg(peers)
	}
}
1352503
func (d *driver) createVolumeEnsureAvailabilityZone(
	opts *types.VolumeCreateOpts, server *awsec2.Instance) {

	if opts.AvailabilityZone == nil || *opts.AvailabilityZone == "" {
		opts.AvailabilityZone = server.Placement.AvailabilityZone
	}
}
534793
func (v *azureVolumeSource) createManagedDiskVolumes(ctx context.ProviderCallContext, params []storage.VolumeParams, results []storage.CreateVolumesResult) {
	for i, p := range params {
		if results[i].Error != nil {
			continue
		}
		volume, err := v.createManagedDiskVolume(ctx, p)
		if err != nil {
			results[i].Error = err
			continue
		}
		results[i].Volume = volume
	}
}