console_scripts = cinder-api = cinder.cmd.api:maincinder-backup = cinder.cmd.backup:maincinder-manage = cinder.cmd.manage:maincinder-rootwrap = oslo_rootwrap.cmd:maincinder-rtstool = cinder.cmd.rtstool:maincinder-scheduler = cinder.cmd.scheduler:maincinder-volume = cinder.cmd.volume:maincinder-volume-usage-audit = cinder.cmd.volume_usage_audit:main

def main():objects.register_all()gmr_opts.set_defaults(CONF)CONF(sys.argv[1:], project='cinder',version=version.version_string())config.set_middleware_defaults()logging.setup(CONF, "cinder")python_logging.captureWarnings(True)utils.monkey_patch()gmr.TextGuruMeditation.setup_autorun(version, conf=CONF)rpc.init(CONF)launcher = service.process_launcher()server = service.WSGIService('osapi_volume')launcher.launch_service(server, workers=server.workers)launcher.wait()

def create(self, req, body):...# 对卷的参数进行检查，比如卷名、size参数，参数赋予kwargs字典等
# 调用/cinder/volume/api.py里的create函数进行创建new_volume = self.volume_api.create(context,size,volume.get('display_name'),volume.get('display_description'),**kwargs)retval = self._view_builder.detail(req, new_volume)return retval

def create(self, context, size, name, description, ...):# 验证该用户是否有权限在该project下做这个操作check_policy(context, 'create_from_image' if image_id else 'create')# 获取调用cinder-scheduler服务端方法的rpc客户端sched_rpcapi = (self.scheduler_rpcapi if (not cgsnapshot and not source_cg andnot group_snapshot and not source_group)else None)# 获取调用cinder-volume服务端方法的rpc客户端volume_rpcapi = (self.volume_rpcapi if (not cgsnapshot and not source_cg andnot group_snapshot and not source_group)else None)# 调用/cinder/volume/flows/api/create_volume.py的get_flow方法，并返回api入口流flow_engine = create_volume.get_flow(self.db,self.image_service,availability_zones,create_what,sched_rpcapi,volume_rpcapi)

def get_flow(db_api, image_service_api, availability_zones, create_what,scheduler_rpcapi=None, volume_rpcapi=None):"""Constructs and returns the api entrypoint flow.This flow will do the following:1. Inject keys & values for dependent tasks.2. Extracts and validates the input keys & values.3. Reserves the quota (reverts quota on any failures).4. Creates the database entry.5. Commits the quota.6. Casts to volume manager or scheduler for further processing."""flow_name = ACTION.replace(":", "_") + "_api"# 获取类flow的实例化对象，可以添加task到flow对象api_flow = linear_flow.Flow(flow_name)# 添加ExtractVolumeRequestTask，该任务是将接收到的一系列参数值通过一系列条件的验证将验证结果存储起来供其它task使用
    api_flow.add(ExtractVolumeRequestTask(image_service_api,availability_zones,rebind={'size': 'raw_size','availability_zone': 'raw_availability_zone','volume_type': 'raw_volume_type'}))# 添加多个task# QuotaReserveTask: 为卷保留配额，失败时可以回滚# EntryCreateTask: 将卷的创建过程写入数据库，比如卷的状态，此时卷的状态未"creating"# QuotaCommitTask: 用于提交保留
    api_flow.add(QuotaReserveTask(),EntryCreateTask(),QuotaCommitTask())# 将请求通过基于消息队列服务的rpc方式发送到scheduler或者volume管理程序去处理请求if scheduler_rpcapi and volume_rpcapi:# This will cast it out to either the scheduler or volume manager via# the rpc apis provided.# 添加VolumeCastTask# VolumeCastTask: 将卷创建工作转移到scheduler或volume管理程序去处理，也就表示工作流程从api服务中转到# 其它服务中去执行
        api_flow.add(VolumeCastTask(scheduler_rpcapi, volume_rpcapi, db_api))# Now load (but do not run) the flow using the provided initial data.return taskflow.engines.load(api_flow, store=create_what)

def _cast_create_volume(self, context, request_spec, filter_properties):# 这里判断用户是否有指定host进行创建卷操作，如果没有则将创建任务交给                scheduler管理程序去完成# 如果用户有指定host则跳过scheduler，直接将创建任务交给volume管理程序去完成if not source_volume_ref:# Cast to the scheduler and let it handle whatever is needed# to select the target host for this volume.# 这里会直接调用到SchedulerAPI.create_volume函数# SchedulerAPI.create_volume函数会通过消息异步调用SchedulerManager.create_volume函数，# 也就是/cinder/scheduler/manager.py中的create_volume函数
        self.scheduler_rpcapi.create_volume(context,volume,snapshot_id=snapshot_id,image_id=image_id,request_spec=request_spec,filter_properties=filter_properties)else:# Bypass the scheduler and send the request directly to the volume# manager.volume.host = source_volume_ref.hostvolume.cluster_name = source_volume_ref.cluster_namevolume.scheduled_at = timeutils.utcnow()volume.save()if not cgsnapshot_id:self.volume_rpcapi.create_volume(context,volume,request_spec,filter_properties,allow_reschedule=False)

def create_volume(self, context, volume, snapshot_id=None, image_id=None,request_spec=None, filter_properties=None):# 确保调度程序已经准备好
    self._wait_for_scheduler()try:# 这里会调用/cinder/scheduler/flows/create_volume.py中的get_flow函数flow_engine = create_volume.get_flow(context,self.driver,request_spec,filter_properties,volume,snapshot_id,image_id)

def get_flow(context, driver_api, request_spec=None,filter_properties=None,volume=None, snapshot_id=None, image_id=None):"""Constructs and returns the scheduler entrypoint flow.This flow will do the following:1. Inject keys & values for dependent tasks.2. Extract a scheduler specification from the provided inputs.3. Use provided scheduler driver to select host and pass volume creationrequest further."""create_what = {'context': context,'raw_request_spec': request_spec,'filter_properties': filter_properties,'volume': volume,'snapshot_id': snapshot_id,'image_id': image_id,}flow_name = ACTION.replace(":", "_") + "_scheduler"# 获取类flow实例对象scheduler_flow = linear_flow.Flow(flow_name)# This will extract and clean the spec from the starting values.# ExtractSchedulerSpecTask: 从请求规范中提取规范对象
    scheduler_flow.add(ExtractSchedulerSpecTask(rebind={'request_spec': 'raw_request_spec'}))# This will activate the desired scheduler driver (and handle any# driver related failures appropriately).# ScheduleCreateVolumeTask: 激活scheduler程序的驱动程序并处理任何后续故障
    scheduler_flow.add(ScheduleCreateVolumeTask(driver_api))# Now load (but do not run) the flow using the provided initial data.return taskflow.engines.load(scheduler_flow, store=create_what)

def schedule_create_volume(self, context, request_spec, filter_properties):# 选择一个最合适的backend来进行准备创建的卷的存储backend = self._schedule(context, request_spec, filter_properties)if not backend:raise exception.NoValidBackend(reason=_("No weighed backends ""available"))backend = backend.objvolume_id = request_spec['volume_id']# 更新数据库当前卷的状态updated_volume = driver.volume_update_db(context, volume_id,backend.host,backend.cluster_name)self._post_select_populate_filter_properties(filter_properties,backend)# context is not serializablefilter_properties.pop('context', None)# 通过消息队列请求调用volume_rpcapi.create_volume# VolumeAPI.create_volume会通过消息队列远程调用VolumeManager.create_volume# 最后会调用到/cinder/volume/manager.py中的create_volume函数，也就是创建卷的工作流程会转入到cinder-volume的服务中
    self.volume_rpcapi.create_volume(context, updated_volume, request_spec,filter_properties,allow_reschedule=True)

def get_flow(context, manager, db, driver, scheduler_rpcapi, host, volume,allow_reschedule, reschedule_context, request_spec,filter_properties, image_volume_cache=None):"""Constructs and returns the manager entrypoint flow.This flow will do the following:1. Determines if rescheduling is enabled (ahead of time).2. Inject keys & values for dependent tasks.3. Selects 1 of 2 activated only on *failure* tasks (one to update the dbstatus & notify or one to update the db status & notify & *reschedule*).4. Extracts a volume specification from the provided inputs.5. Notifies that the volume has started to be created.6. Creates a volume from the extracted volume specification.7. Attaches a on-success *only* task that notifies that the volume creationhas ended and performs further database status updates."""flow_name = ACTION.replace(":", "_") + "_manager"# 获取类flow实例对象volume_flow = linear_flow.Flow(flow_name)# This injects the initial starting flow values into the workflow so that# the dependency order of the tasks provides/requires can be correctly# determined.create_what = {'context': context,'filter_properties': filter_properties,'request_spec': request_spec,'volume': volume,}# ExtractVolumeRefTask: 提取给定卷ID的卷引用volume_flow.add(ExtractVolumeRefTask(db, host, set_error=False))retry = filter_properties.get('retry', None)# Always add OnFailureRescheduleTask and we handle the change of volume's# status when reverting the flow. Meanwhile, no need to revert process of# ExtractVolumeRefTask.do_reschedule = allow_reschedule and request_spec and retryvolume_flow.add(OnFailureRescheduleTask(reschedule_context, db, driver,scheduler_rpcapi, do_reschedule))LOG.debug("Volume reschedule parameters: %(allow)s ""retry: %(retry)s", {'allow': allow_reschedule, 'retry': retry})# ExtractVolumeSpecTask: 结合数据库存取的该卷的信息，提供有用的、易分析的卷相关数据结构给其它任务使用# NotifyVolumeActionTask: 在卷开始创建时执行相关的通知信息# CreateVolumeFromSpecTask: 该任务是根据卷的规格真实创建卷# CreateVolumeOnFinishTask: 当卷创建成功后，会使用该任务将卷在数据库的状态更新为available
    volume_flow.add(ExtractVolumeSpecTask(db),NotifyVolumeActionTask(db, "create.start"),CreateVolumeFromSpecTask(manager,db,driver,image_volume_cache),CreateVolumeOnFinishTask(db, "create.end"))# Now load (but do not run) the flow using the provided initial data.return taskflow.engines.load(volume_flow, store=create_what)

# 根据不同类型的卷调用不同的方法来创建
if create_type == 'raw':model_update = self._create_raw_volume(volume, **volume_spec)
elif create_type == 'snap':model_update = self._create_from_snapshot(context, volume,**volume_spec)
elif create_type == 'source_vol':model_update = self._create_from_source_volume(context, volume, **volume_spec)
elif create_type == 'source_replica':model_update = self._create_from_source_replica(context, volume, **volume_spec)

ret = self.driver.create_volume(volume)

def create(self, req, server_id, body):"""Attach a volume to an instance."""context = req.environ['nova.context']context.can(vol_policies.BASE_POLICY_NAME)context.can(va_policies.POLICY_ROOT % 'create')volume_id = body['volumeAttachment']['volumeId']device = body['volumeAttachment'].get('device')instance = common.get_instance(self.compute_api, context, server_id)if instance.vm_state in (vm_states.SHELVED,vm_states.SHELVED_OFFLOADED):_check_request_version(req, '2.20', 'attach_volume',server_id, instance.vm_state)device = self.compute_api.attach_volume(context, instance,volume_id, device)# The attach is asyncattachment = {}attachment['id'] = volume_idattachment['serverId'] = server_idattachment['volumeId'] = volume_idattachment['device'] = device# TODO(justinsb): How do I return "accepted" here?return {'volumeAttachment': attachment}

def attach_volume(self, context, instance, volume_id, device=None,disk_bus=None, device_type=None):if device and not block_device.match_device(device):raise exception.InvalidDevicePath(path=device)is_shelved_offloaded = instance.vm_state == vm_states.SHELVED_OFFLOADEDif is_shelved_offloaded:return self._attach_volume_shelved_offloaded(context,instance,volume_id,device,disk_bus,device_type)return self._attach_volume(context, instance, volume_id, device,disk_bus, device_type)

def _attach_volume(self, context, instance, volume_id, device,disk_bus, device_type):# 远程目的主机确定设备名和更新数据库并返回相关信息对象volume_bdm = self._create_volume_bdm(context, instance, device, volume_id, disk_bus=disk_bus,device_type=device_type)try:# 这个函数中会通过远程调用cinder-volume服务里的函数来检查该卷是否是可添加的且更新数据库的状态
        self._check_attach_and_reserve_volume(context, volume_id, instance)self.compute_rpcapi.attach_volume(context, instance, volume_bdm)except Exception:with excutils.save_and_reraise_exception():volume_bdm.destroy()return volume_bdm.device_name

def _attach_volume(self, context, instance, bdm):context = context.elevated()# 因为我们创建的volume，所以这里调用的attach方法是/nova/virt/block_device.py里的attach函数
    bdm.attach(context, instance, self.volume_api, self.driver,do_check_attach=False, do_driver_attach=True)info = {'volume_id': bdm.volume_id}self._notify_about_instance_usage(context, instance, "volume.attach", extra_usage_info=info)

def attach(self, context, instance, volume_api, virt_driver,do_check_attach=True, do_driver_attach=False, **kwargs):# 获取有关要attach的volume的信息对象volume = volume_api.get(context, self.volume_id)if do_check_attach:# 检查volume的状态属性是否符合attach的要求volume_api.check_attach(context, volume, instance=instance)volume_id = volume['id']context = context.elevated()# 返回该计算节点信息，比如节点名、节点ip、操作系统架构等connector = virt_driver.get_volume_connector(instance)# 获取卷的信息，比如如果是存储在Ceph集群的，会包含集群名、monitor节点ip等，确保拥有的信息能访问集群中的该volumeconnection_info = volume_api.initialize_connection(context,volume_id,connector)if 'serial' not in connection_info:connection_info['serial'] = self.volume_idself._preserve_multipath_id(connection_info)# If do_driver_attach is False, we will attach a volume to an instance# at boot time. So actual attach is done by instance creation code.if do_driver_attach:# 远程调用cinder组件服务中的函数去获取该卷的加密元数据encryption = encryptors.get_encryption_metadata(context, volume_api, volume_id, connection_info)virt_driver.attach_volume(context, connection_info, instance,self['mount_device'], disk_bus=self['disk_bus'],device_type=self['device_type'], encryption=encryption)self['connection_info'] = connection_infoif self.volume_size is None:self.volume_size = volume.get('size')mode = 'rw'if 'data' in connection_info:mode = connection_info['data'].get('access_mode', 'rw')

def initialize_connection(self, context, volume, connector):# TODO(jdg): Add deprecation warning# 验证driver是否已经初始化
    utils.require_driver_initialized(self.driver)# 对于rbd驱动没有重写该方法，所以没做任何事
    self.driver.validate_connector(connector)# 对于rbd driver，该方法是空的，因为rbd不需要像ISCSI那样创建target、创建protalmodel_update = self.driver.create_export(context.elevated(),volume, connector)if model_update:volume.update(model_update)volume.save()# 对于ceph rbd驱动，这里是获取有关该卷所在集群的信息，比如monitor、ip、secret_uuid等conn_info = self.driver.initialize_connection(volume, connector)conn_info = self._parse_connection_options(context, volume, conn_info)LOG.info(_LI("Initialize volume connection completed successfully."),resource=volume)
return conn_info

def attach_volume(self, context, connection_info, instance, mountpoint,disk_bus=None, device_type=None, encryption=None):guest = self._host.get_guest(instance)disk_dev = mountpoint.rpartition("/")[2]bdm = {'device_name': disk_dev,'disk_bus': disk_bus,'device_type': device_type}# Note(cfb): If the volume has a custom block size, check that#            that we are using QEMU/KVM and libvirt >= 0.10.2. The#            presence of a block size is considered mandatory by#            cinder so we fail if we can't honor the request.data = {}if ('data' in connection_info):data = connection_info['data']if ('logical_block_size' in data or 'physical_block_size' in data):if ((CONF.libvirt.virt_type != "kvm" andCONF.libvirt.virt_type != "qemu")):msg = _("Volume sets block size, but the current ""libvirt hypervisor '%s' does not support custom ""block size") % CONF.libvirt.virt_typeraise exception.InvalidHypervisorType(msg)disk_info = blockinfo.get_info_from_bdm(instance, CONF.libvirt.virt_type, instance.image_meta, bdm)self._connect_volume(connection_info, disk_info)if disk_info['bus'] == 'scsi':disk_info['unit'] = self._get_scsi_controller_max_unit(guest) + 1conf = self._get_volume_config(connection_info, disk_info)self._check_discard_for_attach_volume(conf, instance)state = guest.get_power_state(self._host)live = state in (power_state.RUNNING, power_state.PAUSED)if encryption:encryptor = self._get_volume_encryptor(connection_info,encryption)encryptor.attach_volume(context, **encryption)guest.attach_device(conf, persistent=True, live=live)

def attach_device(self, conf, persistent=False, live=False):"""Attaches device to the guest.:param conf: A LibvirtConfigObject of the device to attach:param persistent: A bool to indicate whether the change ispersistent or not:param live: A bool to indicate whether it affect the guestin running state"""flags = persistent and libvirt.VIR_DOMAIN_AFFECT_CONFIG or 0flags |= live and libvirt.VIR_DOMAIN_AFFECT_LIVE or 0# 把conf中的信息转换成xml的格式，然后可通过libvirt工具将卷挂载到guest中device_xml = conf.to_xml()if six.PY3 and isinstance(device_xml, six.binary_type):device_xml = device_xml.decode('utf-8')LOG.debug("attach device xml: %s", device_xml)self._domain.attachDeviceFlags(device_xml, flags=flags)