Openstack Nova(九)----Instance 创建（Computer API & Conductor ），openstacknovaapi

Openstack Nova(九)----Instance 创建（Computer API & Conductor ），openstacknovaapi

在前一章中， 学习了所有的流水线处理。最后通过路由的设置，会把创建Instance的请求映射到servers.py的Controller的create方法。这中间会有不少的代码， 但都是一些常用的逻辑， 这里不进行讲述。

Controller/Create

这是一段比较长的代码， 但是做的事情还是比较简单。就是对各种参数的合法性检查。

@wsgi.response(202)
    @wsgi.serializers(xml=FullServerTemplate)
    @wsgi.deserializers(xml=CreateDeserializer)
    def create(self, req, body):
        """Creates a new server for a given user."""
        if not self.is_valid_body(body, 'server'):
            raise exc.HTTPUnprocessableEntity()
        #由前面的流水线生成的上下文环境
        context = req.environ['nova.context']
        #server的全部参数
        server_dict = body['server']
        #Instance的password
        password = self._get_server_admin_password(server_dict)
        #Instance的名称，这个不能为空
        if 'name' not in server_dict:
            msg = _("Server name is not defined")
            raise exc.HTTPBadRequest(explanation=msg)

        name = server_dict['name']
        #验证name的长度1-255
        self._validate_server_name(name)
        name = name.strip()
        #从当前参数中，取出image标识，有可能为空
        image_uuid = self._image_from_req_data(body)

        personality = server_dict.get('personality')
        config_drive = None
        if self.ext_mgr.is_loaded('os-config-drive'):
            config_drive = server_dict.get('config_drive')

        injected_files = []
        if personality:
            #需要注入的文件
            injected_files = self._get_injected_files(personality)

        #安全组配置
        sg_names = []
        if self.ext_mgr.is_loaded('os-security-groups'):
            security_groups = server_dict.get('security_groups')
            if security_groups is not None:
                sg_names = [sg['name'] for sg in security_groups
                            if sg.get('name')]
        if not sg_names:
            sg_names.append('default')

        sg_names = list(set(sg_names))
        #网络配置
        requested_networks = None
        if (self.ext_mgr.is_loaded('os-networks')
                or utils.is_neutron()):
            requested_networks = server_dict.get('networks')

        if requested_networks is not None:
            if not isinstance(requested_networks, list):
                expl = _('Bad networks format')
                raise exc.HTTPBadRequest(explanation=expl)
            requested_networks = self._get_requested_networks(
                requested_networks)
        #ipv4及有效性检查，只是针对IP地址本身，不会考虑其它单元的IP地址
        (access_ip_v4, ) = server_dict.get('accessIPv4'),
        if access_ip_v4 is not None:
            self._validate_access_ipv4(access_ip_v4)
        #ipv6及有效性检查，只是针对IP地址本身，不会考虑其它单元的IP地址
        (access_ip_v6, ) = server_dict.get('accessIPv6'),
        if access_ip_v6 is not None:
            self._validate_access_ipv6(access_ip_v6)

        try:
            #Instance的配置信息， 这个必须有
            flavor_id = self._flavor_id_from_req_data(body)
        except ValueError as error:
            msg = _("Invalid flavorRef provided.")
            raise exc.HTTPBadRequest(explanation=msg)

        # optional openstack extensions:
        key_name = None
        if self.ext_mgr.is_loaded('os-keypairs'):
            key_name = server_dict.get('key_name')

        user_data = None
        if self.ext_mgr.is_loaded('os-user-data'):
            user_data = server_dict.get('user_data')
        self._validate_user_data(user_data)
        #zone信息， 用来对Instance进行分区，分区基于HOST
        availability_zone = None
        if self.ext_mgr.is_loaded('os-availability-zone'):
            availability_zone = server_dict.get('availability_zone')
        #硬盘信息， 可以做为启动盘
        block_device_mapping = None
        block_device_mapping_v2 = None
        legacy_bdm = True
        if self.ext_mgr.is_loaded('os-volumes'):
            block_device_mapping = server_dict.get('block_device_mapping', [])
            for bdm in block_device_mapping:
                try:
                    block_device.validate_device_name(bdm.get("device_name"))
                    block_device.validate_and_default_volume_size(bdm)
                except exception.InvalidBDMFormat as e:
                    raise exc.HTTPBadRequest(explanation=e.format_message())

                if 'delete_on_termination' in bdm:
                    bdm['delete_on_termination'] = strutils.bool_from_string(
                        bdm['delete_on_termination'])

            if self.ext_mgr.is_loaded('os-block-device-mapping-v2-boot'):
                # Consider the new data format for block device mapping
                block_device_mapping_v2 = server_dict.get(
                    'block_device_mapping_v2', [])
                # NOTE (ndipanov):  Disable usage of both legacy and new
                #                   block device format in the same request
                if block_device_mapping and block_device_mapping_v2:
                    expl = _('Using different block_device_mapping syntaxes '
                             'is not allowed in the same request.')
                    raise exc.HTTPBadRequest(explanation=expl)

                # Assume legacy format
                legacy_bdm = not bool(block_device_mapping_v2)

                try:
                    block_device_mapping_v2 = [
                        block_device.BlockDeviceDict.from_api(bdm_dict)
                        for bdm_dict in block_device_mapping_v2]
                except exception.InvalidBDMFormat as e:
                    raise exc.HTTPBadRequest(explanation=e.format_message())

        block_device_mapping = (block_device_mapping or
                                block_device_mapping_v2)

        ret_resv_id = False
        # min_count and max_count are optional.  If they exist, they may come
        # in as strings.  Verify that they are valid integers and > 0.
        # Also, we want to default 'min_count' to 1, and default
        # 'max_count' to be 'min_count'.
        min_count = 1
        max_count = 1
        if self.ext_mgr.is_loaded('os-multiple-create'):
            ret_resv_id = server_dict.get('return_reservation_id', False)
            min_count = server_dict.get('min_count', 1)
            max_count = server_dict.get('max_count', min_count)

        try:
            min_count = utils.validate_integer(
                min_count, "min_count", min_value=1)
            max_count = utils.validate_integer(
                max_count, "max_count", min_value=1)
        except exception.InvalidInput as e:
            raise exc.HTTPBadRequest(explanation=e.format_message())

        if min_count > max_count:
            msg = _('min_count must be <= max_count')
            raise exc.HTTPBadRequest(explanation=msg)

        auto_disk_config = False
        if self.ext_mgr.is_loaded('OS-DCF'):
            auto_disk_config = server_dict.get('auto_disk_config')

        scheduler_hints = {}
        if self.ext_mgr.is_loaded('OS-SCH-HNT'):
            scheduler_hints = server_dict.get('scheduler_hints', {})

        try:
            _get_inst_type = flavors.get_flavor_by_flavor_id
            inst_type = _get_inst_type(flavor_id, ctxt=context,
                                       read_deleted="no")
#使用compute_api去创建Instance
            (instances, resv_id) = self.compute_api.create(context,
                            inst_type,
                            image_uuid,
                            display_name=name,
                            display_description=name,
                            key_name=key_name,
                            metadata=server_dict.get('metadata', {}),
                            access_ip_v4=access_ip_v4,
                            access_ip_v6=access_ip_v6,
                            injected_files=injected_files,
                            admin_password=password,
                            min_count=min_count,
                            max_count=max_count,
                            requested_networks=requested_networks,
                            security_group=sg_names,
                            user_data=user_data,
                            availability_zone=availability_zone,
                            config_drive=config_drive,
                            block_device_mapping=block_device_mapping,
                            auto_disk_config=auto_disk_config,
                            scheduler_hints=scheduler_hints,
                            legacy_bdm=legacy_bdm)
        except exception.QuotaError as error:
            raise exc.HTTPRequestEntityTooLarge(
                explanation=error.format_message(),
                headers={'Retry-After': 0})
        except exception.InvalidMetadataSize as error:
            raise exc.HTTPRequestEntityTooLarge(
                explanation=error.format_message())
        except exception.ImageNotFound as error:
            msg = _("Can not find requested image")
            raise exc.HTTPBadRequest(explanation=msg)
        except exception.FlavorNotFound as error:
            msg = _("Invalid flavorRef provided.")
            raise exc.HTTPBadRequest(explanation=msg)
        except exception.KeypairNotFound as error:
            msg = _("Invalid key_name provided.")
            raise exc.HTTPBadRequest(explanation=msg)
        except exception.ConfigDriveInvalidValue:
            msg = _("Invalid config_drive provided.")
            raise exc.HTTPBadRequest(explanation=msg)
        except rpc_common.RemoteError as err:
            msg = "%(err_type)s: %(err_msg)s" % {'err_type': err.exc_type,
                                                 'err_msg': err.value}
            raise exc.HTTPBadRequest(explanation=msg)
        except UnicodeDecodeError as error:
            msg = "UnicodeError: %s" % unicode(error)
            raise exc.HTTPBadRequest(explanation=msg)
        except (exception.ImageNotActive,
                exception.InstanceTypeDiskTooSmall,
                exception.InstanceTypeMemoryTooSmall,
                exception.InstanceTypeNotFound,
                exception.InvalidMetadata,
                exception.InvalidRequest,
                exception.MultiplePortsNotApplicable,
                exception.PortNotFound,
                exception.SecurityGroupNotFound,
                exception.InvalidBDM) as error:
            raise exc.HTTPBadRequest(explanation=error.format_message())
        except exception.PortInUse as error:
            raise exc.HTTPConflict(explanation=error.format_message())

        # If the caller wanted a reservation_id, return it
        if ret_resv_id:
            return wsgi.ResponseObject({'reservation_id': resv_id},
                                       xml=ServerMultipleCreateTemplate)

        req.cache_db_instances(instances)
        server = self._view_builder.create(req, instances[0])

        if CONF.enable_instance_password:
            server['server']['adminPass'] = password

        robj = wsgi.ResponseObject(server)

        return self._add_location(robj)

compute_api

为了简化问题， 这里假设系统中没有使用cells, 那么对应的computer_api就是 nova/compute/api.py的API类。

create

@hooks.add_hook("create_instance")
    def create(self, context, instance_type,
               image_href, kernel_id=None, ramdisk_id=None,
               min_count=None, max_count=None,
               display_name=None, display_description=None,
               key_name=None, key_data=None, security_group=None,
               availability_zone=None, user_data=None, metadata=None,
               injected_files=None, admin_password=None,
               block_device_mapping=None, access_ip_v4=None,
               access_ip_v6=None, requested_networks=None, config_drive=None,
               auto_disk_config=None, scheduler_hints=None, legacy_bdm=True):
        """
        Provision instances, sending instance information to the
        scheduler.  The scheduler will determine where the instance(s)
        go and will handle creating the DB entries.

        Returns a tuple of (instances, reservation_id)
        """
        #创建规则检查
        self._check_create_policies(context, availability_zone,
                requested_networks, block_device_mapping)

        if requested_networks and max_count > 1 and utils.is_neutron():
            self._check_multiple_instances_neutron_ports(requested_networks)
        #调用自己的_create_instance
        return self._create_instance(
                               context, instance_type,
                               image_href, kernel_id, ramdisk_id,
                               min_count, max_count,
                               display_name, display_description,
                               key_name, key_data, security_group,
                               availability_zone, user_data, metadata,
                               injected_files, admin_password,
                               access_ip_v4, access_ip_v6,
                               requested_networks, config_drive,
                               block_device_mapping, auto_disk_config,
                               scheduler_hints=scheduler_hints,
                               legacy_bdm=legacy_bdm)

_create_instance

def _create_instance(self, context, instance_type,
               image_href, kernel_id, ramdisk_id,
               min_count, max_count,
               display_name, display_description,
               key_name, key_data, security_groups,
               availability_zone, user_data, metadata,
               injected_files, admin_password,
               access_ip_v4, access_ip_v6,
               requested_networks, config_drive,
               block_device_mapping, auto_disk_config,
               reservation_id=None, scheduler_hints=None,
               legacy_bdm=True):
        """Verify all the input parameters regardless of the provisioning
        strategy being performed and schedule the instance(s) for
        creation.
        """

        # Normalize and setup some parameters
        if reservation_id is None:
            reservation_id = utils.generate_uid('r')
        security_groups = security_groups or ['default']
        min_count = min_count or 1
        max_count = max_count or min_count
        block_device_mapping = block_device_mapping or []
        if not instance_type:
            instance_type = flavors.get_default_flavor()

        if image_href:
            image_id, boot_meta = self._get_image(context, image_href)
        else:
            image_id = None
            boot_meta = {}
            boot_meta['properties'] = \
                self._get_bdm_image_metadata(context,
                    block_device_mapping, legacy_bdm)

        self._check_auto_disk_config(image=boot_meta,
                                     auto_disk_config=auto_disk_config)

        handle_az = self._handle_availability_zone
        availability_zone, forced_host, forced_node = handle_az(context,
                                                            availability_zone)

        base_options = self._validate_and_build_base_options(context,
                instance_type, boot_meta, image_href, image_id, kernel_id,
                ramdisk_id, display_name, display_description,
                key_name, key_data, security_groups, availability_zone,
                forced_host, user_data, metadata, injected_files, access_ip_v4,
                access_ip_v6, requested_networks, config_drive,
                block_device_mapping, auto_disk_config, reservation_id)

        block_device_mapping = self._check_and_transform_bdm(
            base_options, boot_meta, min_count, max_count,
            block_device_mapping, legacy_bdm)
        #创建数据库记录
        instances = self._provision_instances(context, instance_type,
                min_count, max_count, base_options, boot_meta, security_groups,
                block_device_mapping)
        #创建过滤条件
        filter_properties = self._build_filter_properties(context,
                scheduler_hints, forced_host, forced_node, instance_type)
        #更新Instance状态
        for instance in instances:
            self._record_action_start(context, instance,
                                      instance_actions.CREATE)
        #使用compute_task_api 去创建Instance
        self.compute_task_api.build_instances(context,
                instances=instances, image=boot_meta,
                filter_properties=filter_properties,
                admin_password=admin_password,
                injected_files=injected_files,
                requested_networks=requested_networks,
                security_groups=security_groups,
                block_device_mapping=block_device_mapping,
                legacy_bdm=False)

        return (instances, reservation_id)

compute_task_api（conductor）/build_instances

def build_instances(self, context, instances, image, filter_properties,
            admin_password, injected_files, requested_networks,
            security_groups, block_device_mapping, legacy_bdm=True):
        self.conductor_compute_rpcapi.build_instances(context,
                instances=instances, image=image,
                filter_properties=filter_properties,
                admin_password=admin_password, injected_files=injected_files,
                requested_networks=requested_networks,
                security_groups=security_groups,
                block_device_mapping=block_device_mapping,
                legacy_bdm=legacy_bdm)

conductor_compute_rpcapi(ComputeTaskAPI)/build_instances

def build_instances(self, context, instances, image, filter_properties,
            admin_password, injected_files, requested_networks,
            security_groups, block_device_mapping, legacy_bdm=True):
        self.conductor_compute_rpcapi.build_instances(context,
                instances=instances, image=image,
                filter_properties=filter_properties,
                admin_password=admin_password, injected_files=injected_files,
                requested_networks=requested_networks,
                security_groups=security_groups,
                block_device_mapping=block_device_mapping,
                legacy_bdm=legacy_bdm)

conductor_compute_rpcapi(conductor/rpcapi/ConductorAPI)

def build_instances(self, context, instances, image, filter_properties,
            admin_password, injected_files, requested_networks,
            security_groups, block_device_mapping, legacy_bdm=True):
        instances_p = [jsonutils.to_primitive(inst) for inst in instances]
        image_p = jsonutils.to_primitive(image)
        cctxt = self.client.prepare(version='1.5')
        cctxt.cast(context, 'build_instances',
                   instances=instances_p, image=image_p,
                   filter_properties=filter_properties,
                   admin_password=admin_password,
                   injected_files=injected_files,
                   requested_networks=requested_networks,
                   security_groups=security_groups,
                   block_device_mapping=block_device_mapping,
                   legacy_bdm=legacy_bdm)

至此， Conductor发送RPC消息给端的Conductor， 然后再交给Scheduler去处理。

但是这里还有一个比较重要的过程，就是RPC机制，下一章中将继续处理。