openstack cinder 创建volume源码分析

曾经终败给现在 2023-10-11 09:58 24阅读 0赞

cinder volume

备注：  
为方便源码阅读， 列出的部分的源码将被省略， 如需要阅读完整代码， 建议在官网进行下载

cinder实现相对比较简单  
基本结构：

1.  api --> volume API --> db
2.  api --> volume API --> rpc --> volume manager --> drivers

--------------------

##### API 入口 #####

目前cinder 已经更新到v3版本， 因此源码阅读的时候， api入口从v3开始  
volume卷 api定义在 cinder/api/v3/volumes.py

class VolumeController(volumes_v2.VolumeController):
        """The Volumes API controller for the OpenStack API V3."""

v3继承v2的功能

目前v3已经更新了v2版本的create  
源码如下

@wsgi.response(202)
        def create(self, req, body):
            """Creates a new volume.
    
            :param req: the request
            :param body: the request body
            :returns: dict -- the new volume dictionary
            :raises: HTTPNotFound, HTTPBadRequest
            """
            self.assert_valid_body(body, 'volume')
    
            LOG.debug('Create volume request body: %s', body)
            context = req.environ['cinder.context']
    
    # 获取并校验参数
            req_version = req.api_version_request
            # Remove group_id from body if max version is less than 3.13.
    				....
    				
            volume = body['volume']
            kwargs = {}
            self.validate_name_and_description(volume)
    
            # NOTE(thingee): v2 API allows name instead of display_name
            if 'name' in volume:
                volume['display_name'] = volume.pop('name')
    
            # NOTE(thingee): v2 API allows description instead of
            #                display_description
            if 'description' in volume:
                volume['display_description'] = volume.pop('description')
    
    # image镜像进行创建
            if 'image_id' in volume:
                volume['imageRef'] = volume.pop('image_id')
    
    # volume 类型
            req_volume_type = volume.get('volume_type', None)
            if req_volume_type:
                # Not found exception will be handled at the wsgi level
                if not uuidutils.is_uuid_like(req_volume_type):
                    kwargs['volume_type'] = (
                        volume_types.get_volume_type_by_name(
                            context, req_volume_type))
                else:
                    kwargs['volume_type'] = volume_types.get_volume_type(
                        context, req_volume_type)
    
            kwargs['metadata'] = volume.get('metadata', None)
    
    
    # 从快照创建
            snapshot_id = volume.get('snapshot_id')
            if snapshot_id is not None:
                # Not found exception will be handled at the wsgi level
                kwargs['snapshot'] = self.volume_api.get_snapshot(context,
                                                                  snapshot_id)
            else:
                kwargs['snapshot'] = None
    
            source_volid = volume.get('source_volid')
            if source_volid is not None:
                # Not found exception will be handled at the wsgi level
                kwargs['source_volume'] = (
                    self.volume_api.get_volume(context,
                                               source_volid))
            else:
                kwargs['source_volume'] = None
    
            source_replica = volume.get('source_replica')
            if source_replica is not None:
                # Not found exception will be handled at the wsgi level
                src_vol = self.volume_api.get_volume(context,
                                                     source_replica)
                if src_vol['replication_status'] == 'disabled':
                    explanation = _('source volume id:%s is not'
                                    ' replicated') % source_replica
                    raise exc.HTTPBadRequest(explanation=explanation)
                kwargs['source_replica'] = src_vol
            else:
                kwargs['source_replica'] = None
    
            consistencygroup_id = volume.get('consistencygroup_id')
            if consistencygroup_id is not None:
                # Not found exception will be handled at the wsgi level
                kwargs['consistencygroup'] = (
                    self.consistencygroup_api.get(context,
                                                  consistencygroup_id))
            else:
                kwargs['consistencygroup'] = None
    
            # Get group_id if volume is in a group.
            group_id = volume.get('group_id')
            if group_id is not None:
                try:
                    kwargs['group'] = self.group_api.get(context, group_id)
                except exception.GroupNotFound as error:
                    raise exc.HTTPNotFound(explanation=error.msg)
    
            size = volume.get('size', None)
            if size is None and kwargs['snapshot'] is not None:
                size = kwargs['snapshot']['volume_size']
            elif size is None and kwargs['source_volume'] is not None:
                size = kwargs['source_volume']['size']
            elif size is None and kwargs['source_replica'] is not None:
                size = kwargs['source_replica']['size']
    
            LOG.info(_LI("Create volume of %s GB"), size)
    
            if self.ext_mgr.is_loaded('os-image-create'):
                image_ref = volume.get('imageRef')
                if image_ref is not None:
                    image_uuid = self._image_uuid_from_ref(image_ref, context)
                    kwargs['image_id'] = image_uuid
    
    
            kwargs['availability_zone'] = volume.get('availability_zone', None)
            
    # 指定调度参数
            kwargs['scheduler_hints'] = volume.get('scheduler_hints', None)
            multiattach = volume.get('multiattach', False)
            kwargs['multiattach'] = multiattach
    
    # API 创建volume卷
            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

创建volume卷`self.volume_api.create`, 源码定义在cinder/volume/api.py中

def create(self, context, size, name, description, snapshot=None,
                   image_id=None, volume_type=None, metadata=None,
                   availability_zone=None, source_volume=None,
                   scheduler_hints=None,
                   source_replica=None, consistencygroup=None,
                   cgsnapshot=None, multiattach=False, source_cg=None,
                   group=None, group_snapshot=None, source_group=None):
    
            check_policy(context, 'create')
    
    
    # 参数校验， 大小校验， 合规性校验等
        		.....
        		
        		
            if volume_type and 'extra_specs' not in volume_type:
                extra_specs = volume_types.get_volume_type_extra_specs(
                    volume_type['id'])
                volume_type['extra_specs'] = extra_specs
    
            if source_volume and volume_type:
                if volume_type['id'] != source_volume['volume_type_id']:
                    if not self._retype_is_possible(
                            context,
                            volume_type['id'],
                            source_volume['volume_type_id'],
                            volume_type):
                        msg = _("Invalid volume_type provided: %s (requested type "
                                "is not compatible; either match source volume, "
                                "or omit type argument).") % volume_type['id']
                        raise exception.InvalidInput(reason=msg)
    
            # When cloning replica (for testing), volume type must be omitted
            if source_replica and volume_type:
                msg = _("No volume_type should be provided when creating test "
                        "replica.")
                raise exception.InvalidInput(reason=msg)
    
    # 校验快照
            if snapshot and volume_type:
                if volume_type['id'] != snapshot.volume_type_id:
                    if not self._retype_is_possible(context,
                                                    volume_type['id'],
                                                    snapshot.volume_type_id,
                                                    volume_type):
                        msg = _("Invalid volume_type provided: %s (requested "
                                "type is not compatible; recommend omitting "
                                "the type argument).") % volume_type['id']
                        raise exception.InvalidInput(reason=msg)
    
            # Determine the valid availability zones that the volume could be
            # created in (a task in the flow will/can use this information to
            # ensure that the availability zone requested is valid).
            raw_zones = self.list_availability_zones(enable_cache=True)
            availability_zones = set([az['name'] for az in raw_zones])
            if CONF.storage_availability_zone:
                availability_zones.add(CONF.storage_availability_zone)
    
            utils.check_metadata_properties(metadata)
    
    # 设置参数
            create_what = {
                'context': context,
                'raw_size': size,
                'name': name,
                'description': description,
                'snapshot': snapshot,
                'image_id': image_id,
                'raw_volume_type': volume_type,
                'metadata': metadata or {},
                'raw_availability_zone': availability_zone,
                'source_volume': source_volume,
                'scheduler_hints': scheduler_hints,
                'key_manager': self.key_manager,
                'source_replica': source_replica,
                'optional_args': {'is_quota_committed': False},
                'consistencygroup': consistencygroup,
                'cgsnapshot': cgsnapshot,
                'multiattach': multiattach,
                'group': group,
                'group_snapshot': group_snapshot,
                'source_group': source_group,
            }
            
            
    # 设置响应的参数，及初始化
            try:
                sched_rpcapi = (self.scheduler_rpcapi if (
                                not cgsnapshot and not source_cg and
                                not group_snapshot and not source_group)
                                else None)
                volume_rpcapi = (self.volume_rpcapi if (
                                 not cgsnapshot and not source_cg and
                                 not group_snapshot and not source_group)
                                 else None)
                flow_engine = create_volume.get_flow(self.db,
                                                     self.image_service,
                                                     availability_zones,
                                                     create_what,
                                                     sched_rpcapi,
                                                     volume_rpcapi)
            except Exception:
                msg = _('Failed to create api volume flow.')
                LOG.exception(msg)
                raise exception.CinderException(msg)
    
            # Attaching this listener will capture all of the notifications that
            # taskflow sends out and redirect them to a more useful log for
            # cinders debugging (or error reporting) usage.
            with flow_utils.DynamicLogListener(flow_engine, logger=LOG):
    # 执行
                flow_engine.run()
                vref = flow_engine.storage.fetch('volume')
                LOG.info(_LI("Volume created successfully."), resource=vref)
                return vref

上面的create\_volume功能已经被封装：

sched_rpcapi = (self.scheduler_rpcapi if (
                    not cgsnapshot and not source_cg and
                    not group_snapshot and not source_group)
                    else None)
    volume_rpcapi = (self.volume_rpcapi if (
                     not cgsnapshot and not source_cg and
                     not group_snapshot and not source_group)
                     else None)
    flow_engine = create_volume.get_flow(self.db,
                                         self.image_service,
                                         availability_zones,
                                         create_what,
                                         sched_rpcapi,
                                         volume_rpcapi)

具体来看一下的实现， 该函数封装了crate\_volume功能

flow_engine = create_volume.get_flow(self.db,
                                         self.image_service,
                                         availability_zones,
                                         create_what,
                                         sched_rpcapi,
                                         volume_rpcapi)

该方法定义在cinder/volume/flows/api/create\_volume.py 中：

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"
        api_flow = linear_flow.Flow(flow_name)
    
        api_flow.add(ExtractVolumeRequestTask(
            image_service_api,
            availability_zones,
            rebind={'size': 'raw_size',
                    'availability_zone': 'raw_availability_zone',
                    'volume_type': 'raw_volume_type'}))
        api_flow.add(QuotaReserveTask(),
                     EntryCreateTask(db_api),
                     QuotaCommitTask())
    
        if scheduler_rpcapi and volume_rpcapi:
            # This will cast it out to either the scheduler or volume manager via
            # the rpc apis provided.
            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)

api\_flow 可以知道create任务， 实现了scheduler\_rpcapi, volume\_rpcapi, db\_api  
三个任务的任务流调用

具体实现：  
`VolumeCastTask(scheduler_rpcapi, volume_rpcapi, db_api)`

该函数定义通过execute进行调用

定义：

def execute(self, context, **kwargs):
            scheduler_hints = kwargs.pop('scheduler_hints', None)
            db_vt = kwargs.pop('volume_type')
            kwargs['volume_type'] = None
            if db_vt:
                kwargs['volume_type'] = objects.VolumeType()
                objects.VolumeType()._from_db_object(context,
                                                     kwargs['volume_type'], db_vt)
            request_spec = objects.RequestSpec(**kwargs)
            filter_properties = {}
            if scheduler_hints:
                filter_properties['scheduler_hints'] = scheduler_hints
            self._cast_create_volume(context, request_spec, filter_properties)

实际调用在`_cast_create_volume`

def _cast_create_volume(self, context, request_spec, filter_properties):
            source_volid = request_spec['source_volid']
            source_replicaid = request_spec['source_replicaid']
            volume_id = request_spec['volume_id']
            volume = request_spec['volume']
            snapshot_id = request_spec['snapshot_id']
            image_id = request_spec['image_id']
            cgroup_id = request_spec['consistencygroup_id']
            host = None
            cgsnapshot_id = request_spec['cgsnapshot_id']
            group_id = request_spec['group_id']
            if cgroup_id:
                # If cgroup_id existed, we should cast volume to the scheduler
                # to choose a proper pool whose backend is same as CG's backend.
                cgroup = objects.ConsistencyGroup.get_by_id(context, cgroup_id)
                request_spec['CG_backend'] = vol_utils.extract_host(cgroup.host)
            elif group_id:
                # If group_id exists, we should cast volume to the scheduler
                # to choose a proper pool whose backend is same as group's backend.
                group = objects.Group.get_by_id(context, group_id)
                # FIXME(wanghao): group_backend got added before request_spec was
                # converted to versioned objects. We should make sure that this
                # will be handled by object version translations once we add
                # RequestSpec object.
                request_spec['group_backend'] = vol_utils.extract_host(group.host)
            elif snapshot_id and CONF.snapshot_same_host:
                # NOTE(Rongze Zhu): A simple solution for bug 1008866.
                #
                # If snapshot_id is set and CONF.snapshot_same_host is True, make
                # the call create volume directly to the volume host where the
                # snapshot resides instead of passing it through the scheduler, so
                # snapshot can be copied to the new volume.
                snapshot = objects.Snapshot.get_by_id(context, snapshot_id)
                source_volume_ref = objects.Volume.get_by_id(context,
                                                             snapshot.volume_id)
                host = source_volume_ref.host
            elif source_volid:
                source_volume_ref = objects.Volume.get_by_id(context,
                                                             source_volid)
                host = source_volume_ref.host
            elif source_replicaid:
                source_volume_ref = objects.Volume.get_by_id(context,
                                                             source_replicaid)
                host = source_volume_ref.host
    
    如果没有host则进行调度后创建volume
            if not host:
                # Cast to the scheduler and let it handle whatever is needed
                # to select the target host for this volume.
                self.scheduler_rpcapi.create_volume(
                    context,
                    constants.VOLUME_TOPIC,
                    volume_id,
                    snapshot_id=snapshot_id,
                    image_id=image_id,
                    request_spec=request_spec,
                    filter_properties=filter_properties,
                    volume=volume)
            else:
    # 拥有host， 则直接rpc创建volume
                # Bypass the scheduler and send the request directly to the volume
                # manager.
                volume.host = host
                volume.scheduled_at = timeutils.utcnow()
                volume.save()
                if not cgsnapshot_id:
                    self.volume_rpcapi.create_volume(
                        context,
                        volume,
                        volume.host,
                        request_spec,
                        filter_properties,
                        allow_reschedule=False)

没有指定的host作为扩展内容进行分析， 这里对拥有host的进行分析  
volume\_rpcapi.create\_volume调用host的rpc，在远程主机上创建volume卷

--------------------

##### rpc #####

源码实现在cinder/volume/manager.py

def create_volume(self, context, volume_id, request_spec=None,
                          filter_properties=None, allow_reschedule=True,
                          volume=None):
            """Creates the volume."""
            # Log about unsupported drivers
            utils.log_unsupported_driver_warning(self.driver)
    
    # 获取volume信息
            # FIXME(dulek): Remove this in v3.0 of RPC API.
            if volume is None:
                # For older clients, mimic the old behavior and look up the volume
                # by its volume_id.
                volume = objects.Volume.get_by_id(context, volume_id)
    
            # FIXME(dulek): Remove this in v3.0 of RPC API.
            if isinstance(request_spec, dict):
                # We may receive request_spec as dict from older clients.
                request_spec = objects.RequestSpec.from_primitives(request_spec)
    
            context_elevated = context.elevated()
            if filter_properties is None:
                filter_properties = {}
    # 设置request_spec方法
            if request_spec is None:
                request_spec = objects.RequestSpec()
    
            try:
                # NOTE(flaper87): Driver initialization is
                # verified by the task itself.
                flow_engine = create_volume.get_flow(
                    context_elevated,
                    self,
                    self.db,
                    self.driver,
                    self.scheduler_rpcapi,
                    self.host,
                    volume,
                    allow_reschedule,
                    context,
                    request_spec,
                    filter_properties,
                    image_volume_cache=self.image_volume_cache,
                )
            except Exception:
                msg = _("Create manager volume flow failed.")
                LOG.exception(msg, resource={'type': 'volume', 'id': volume.id})
                raise exception.CinderException(msg)
    
            snapshot_id = request_spec.get('snapshot_id')
            source_volid = request_spec.get('source_volid')
            source_replicaid = request_spec.get('source_replicaid')
    
    				......
    
            def _run_flow():
                # This code executes create volume flow. If something goes wrong,
                # flow reverts all job that was done and reraises an exception.
                # Otherwise, all data that was generated by flow becomes available
                # in flow engine's storage.
                with flow_utils.DynamicLogListener(flow_engine, logger=LOG):
                    flow_engine.run()
    
            rescheduled = False
    
            try:
                if locked_action is None:
                    _run_flow()
                else:
                    with coordination.Lock(locked_action):
                        _run_flow()
            finally:
                try:
                    flow_engine.storage.fetch('refreshed')
                except tfe.NotFound:
    					
    				.....
    				
    				
            LOG.info(_LI("Created volume successfully."), resource=volume)
            return volume.id

调用create\_volume.get\_flow 进行run创建， 实际使用excute执行

def get_flow(context, manager, db, driver, scheduler_rpcapi, host, volume,
                 allow_reschedule, reschedule_context, request_spec,
                 filter_properties, image_volume_cache=None):
    
    		....
    		
        do_reschedule = allow_reschedule and request_spec and retry
        volume_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})
    
        volume_flow.add(ExtractVolumeSpecTask(db),
        
    # NOTIFY 任务
                        NotifyVolumeActionTask(db, "create.start"),
    # 创建volume                   
                        CreateVolumeFromSpecTask(manager,
                                                 db,
                                                 driver,
                                                 image_volume_cache),
                                                 
    # 创建volume完成收尾功能                                            
                        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)

CreateVolumeFromSpecTask 调用execute创建volume卷

def execute(self, context, volume, volume_spec):
            volume_spec = dict(volume_spec)
            volume_id = volume_spec.pop('volume_id', None)
            if not volume_id:
                volume_id = volume.id
    
            # we can't do anything if the driver didn't init
            if not self.driver.initialized:
                driver_name = self.driver.__class__.__name__
                LOG.error(_LE("Unable to create volume. "
                              "Volume driver %s not initialized"), driver_name)
                raise exception.DriverNotInitialized()
    
            create_type = volume_spec.pop('type', None)
            LOG.info(_LI("Volume %(volume_id)s: being created as %(create_type)s "
                         "with specification: %(volume_spec)s"),
                     {'volume_spec': volume_spec, 'volume_id': volume_id,
                      'create_type': create_type})
            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)
            elif create_type == 'image':
                model_update = self._create_from_image(context,
                                                       volume,
                                                       **volume_spec)
            else:
                raise exception.VolumeTypeNotFound(volume_type_id=create_type)
    
            # Persist any model information provided on creation.
            try:
                if model_update:
                    volume.update(model_update)
                    volume.save()
            except exception.CinderException:
                # If somehow the update failed we want to ensure that the
                # failure is logged (but not try rescheduling since the volume at
                # this point has been created).
                LOG.exception(_LE("Failed updating model of volume %(volume_id)s "
                                  "with creation provided model %(model)s"),
                              {'volume_id': volume_id, 'model': model_update})
                raise

volume依据不同的创建来源，调用不同方法进行处理

例如raw创建：

if create_type == 'raw':
                model_update = self._create_raw_volume(volume, **volume_spec)

实现：

def _create_raw_volume(self, volume, **kwargs):
            return self.driver.create_volume(volume)

直接调用driver驱动进行创建

--------------------

##### 驱动 #####

驱动定义在cinder/volume/dirvers/ 下：

如果是ceph则使用rbd.py  
实现为：

def create_volume(self, volume):
            """Creates a logical volume."""
    
            if volume.encryption_key_id:
                message = _("Encryption is not yet supported.")
                raise exception.VolumeDriverException(message=message)
    
            size = int(volume.size) * units.Gi
    
            LOG.debug("creating volume '%s'", volume.name)
    
            chunk_size = self.configuration.rbd_store_chunk_size * units.Mi
            order = int(math.log(chunk_size, 2))
    
            with RADOSClient(self) as client:
                self.RBDProxy().create(client.ioctx,
                                       utils.convert_str(volume.name),
                                       size,
                                       order,
                                       old_format=False,
                                       features=client.features)

调用rbd api进行创建

创建完成后由CreateVolumeOnFinishTask 回写volume信息

其他来源与raw的类似， 可以进行参考， 这里不做具体分析

--------------------

##### 扩展： #####

前面进行了host已知的分析， 这里补充host未知的情况下volume create的分析：  
回到cinder/volume/flows/api/create\_volume.py

if not host:
                # Cast to the scheduler and let it handle whatever is needed
                # to select the target host for this volume.
                self.scheduler_rpcapi.create_volume(
                    context,
                    constants.VOLUME_TOPIC,
                    volume_id,
                    snapshot_id=snapshot_id,
                    image_id=image_id,
                    request_spec=request_spec,
                    filter_properties=filter_properties,
                    volume=volume)

scheduler\_rpcapi.create\_volume实现在cinder/scheduler/manager.py

def create_volume(self, context, topic, volume_id, snapshot_id=None,
                          image_id=None, request_spec=None,
                          filter_properties=None, volume=None):
    
            self._wait_for_scheduler()
    
    			.......
    
            try:
                flow_engine = create_volume.get_flow(context,
                                                     db, self.driver,
                                                     request_spec,
                                                     filter_properties,
                                                     volume,
                                                     snapshot_id,
                                                     image_id)
            except Exception:
    				......

`create_volume.get_flow` 实现：

def get_flow(context, db_api, driver_api, request_spec=None,
                 filter_properties=None,
                 volume=None, snapshot_id=None, image_id=None):
    
    		......
    
        # This will activate the desired scheduler driver (and handle any
        # driver related failures appropriately).
        scheduler_flow.add(ScheduleCreateVolumeTask(db_api, driver_api))

任务流实现在： ScheduleCreateVolumeTask execute

def execute(self, context, request_spec, filter_properties, volume):
            try:
                self.driver_api.schedule_create_volume(context, request_spec,
                                                       filter_properties)
            except Exception as e:
    					......

schedule\_create\_volume 定义在cinder/scheduler/filter\_scheduler.py

def schedule_create_volume(self, context, request_spec, filter_properties):
        
    # 调度主机
            weighed_host = self._schedule(context, request_spec,
                                          filter_properties)
    
            if not weighed_host:
                raise exception.NoValidHost(reason=_("No weighed hosts available"))
    
            host = weighed_host.obj.host
            volume_id = request_spec['volume_id']
    
    # 将host信息回写volume db中
            updated_volume = driver.volume_update_db(context, volume_id, host)
            self._post_select_populate_filter_properties(filter_properties,
                                                         weighed_host.obj)
    
            # context is not serializable
            filter_properties.pop('context', None)
    
            self.volume_rpcapi.create_volume(context, updated_volume, host,
                                             request_spec, filter_properties,
                                             allow_reschedule=True)

volume\_rpcapi.create\_volume 调用host的rpc创建volume

之后的流程与之前拥有host的流程一致， 这里不做重复分析