openstack-nova源码分析(七）host aggregates 主机组

╰+攻爆jí腚メ 2022-02-21 12:51 390阅读 0赞

Host aggregates 主机组

主机组是一组主机compute节点的集合，默认只开放给管理员的功能，可以通过策略配置修改访问权限

官方api介绍：

Creates and manages host aggregates. An aggregate assigns metadata to groups of compute nodes.
    Policy defaults enable only users with the administrative role to perform operations with aggregates. 
    Cloud providers can change these permissions through policy file configuration.

#### 一. url及命令 ####

1.  url  
    标准的restful api  
    /os-aggregates  
    /os-aggregates/\{aggregate\_id\}
2.  命令

[root@nova ~]# nova help | grep aggregate
        aggregate-create            Create a new aggregate with the specified
        aggregate-delete            Delete the aggregate.
        aggregate-list              Print a list of all aggregates.
        aggregate-show              Show details of the specified aggregate.
        aggregate-update            Update the aggregate's name and optionally

#### 二. 源码分析： ####

在setup.cfg中定义如下：

aggregates = nova.api.openstack.compute.aggregates:Aggregates

Aggregates定义如下：

class Aggregates(extensions.V21APIExtensionBase):
        """Admin-only aggregate administration."""
    
        name = "Aggregates"
        alias = ALIAS
        version = 1
    
        def get_resources(self):
            resources = [extensions.ResourceExtension(
                                                ALIAS,
                                                AggregateController(),
                                                member_actions={'action': 'POST'})]
            return resources

映射到AggregateController并增加了POST 方法映射到action

1.  列表功能映射到index， 定义如下：

@extensions.expected_errors(())
        def index(self, req):
            """Returns a list a host aggregate's id, name, availability_zone."""
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'index')
            aggregates = self.api.get_aggregate_list(context)
            return {'aggregates': [self._marshall_aggregate(req, a)['aggregate']
                                   for a in aggregates]}

self.api.get\_aggregate\_list(context) 获取列表

实际调用如下：

def get_aggregate_list(self, context):
            """Get all the aggregates."""
            return objects.AggregateList.get_all(context)

调用object层的AggregateList的get\_all方法

1.  创建

@extensions.expected_errors((400, 409))
        @validation.schema(aggregates.create_v20, '2.0', '2.0')
        @validation.schema(aggregates.create, '2.1')
        def create(self, req, body):
            """Creates an aggregate, given its name and
            optional availability zone.
            """
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'create')
            host_aggregate = body["aggregate"]
            name = common.normalize_name(host_aggregate["name"])
            avail_zone = host_aggregate.get("availability_zone")
            if avail_zone:
                avail_zone = common.normalize_name(avail_zone)
    
            try:
                aggregate = self.api.create_aggregate(context, name, avail_zone)
            except exception.AggregateNameExists as e:
                raise exc.HTTPConflict(explanation=e.format_message())
            except exception.ObjectActionError:
                raise exc.HTTPConflict(explanation=_(
                    'Not all aggregates have been migrated to the API database'))
            except exception.InvalidAggregateAction as e:
                raise exc.HTTPBadRequest(explanation=e.format_message())

实际创建：

aggregate = self.api.create\_aggregate(context, name, avail\_zone)

create\_aggregate定义：

def create_aggregate(self, context, aggregate_name, availability_zone):
            """Creates the model for the aggregate."""
    
            aggregate = objects.Aggregate(context=context)
            aggregate.name = aggregate_name
            if availability_zone:
                aggregate.metadata = {'availability_zone': availability_zone}
            aggregate.create()
            self.scheduler_client.update_aggregates(context, [aggregate])
            return aggregate

调用object层的Aggregate类的create方法  
创建完成后self.scheduler\_client.update\_aggregates(context, \[aggregate\]) 刷新nova虚拟机调度器的数据，

def create(self):
            if self.obj_attr_is_set('id'):
                raise exception.ObjectActionError(action='create',
                                                  reason='already created')
    
            # NOTE(mdoff): Once we have made it past a point where we know
            # all aggregates have been migrated, we can remove this. Ideally
            # in Ocata with a blocker migration to be sure.
            if not self._ensure_migrated(self._context):
                raise exception.ObjectActionError(
                    action='create',
                    reason='main database still contains aggregates')
    
            self._assert_no_hosts('create')
            updates = self.obj_get_changes()
            payload = dict(updates)
            if 'metadata' in updates:
                # NOTE(danms): For some reason the notification format is weird
                payload['meta_data'] = payload.pop('metadata')
            if 'uuid' not in updates:
                updates['uuid'] = uuidutils.generate_uuid()
                self.uuid = updates['uuid']
                LOG.debug('Generated uuid %(uuid)s for aggregate',
                          dict(uuid=updates['uuid']))
            compute_utils.notify_about_aggregate_update(self._context,
                                                        "create.start",
                                                        payload)
            compute_utils.notify_about_aggregate_action(
                context=self._context,
                aggregate=self,
                action=fields.NotificationAction.CREATE,
                phase=fields.NotificationPhase.START)
    
            metadata = updates.pop('metadata', None)
            db_aggregate = _aggregate_create_in_db(self._context, updates,
                                                   metadata=metadata)
            self._from_db_object(self._context, self, db_aggregate)
            payload['aggregate_id'] = self.id
            compute_utils.notify_about_aggregate_update(self._context,
                                                        "create.end",
                                                        payload)
            compute_utils.notify_about_aggregate_action(
                context=self._context,
                aggregate=self,
                action=fields.NotificationAction.CREATE,
                phase=fields.NotificationPhase.END)

创建create函数：  
首先\_ensure\_migrated 校验主机是可用状态，不是delete  
\_aggregate\_create\_in\_db创建主机组数据，并在metadata中创建对应的元数据

1.  详情接口，show：

@extensions.expected_errors(404)
        def show(self, req, id):
            """Shows the details of an aggregate, hosts and metadata included."""
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'show')
            try:
                aggregate = self.api.get_aggregate(context, id)
            except exception.AggregateNotFound as e:
                raise exc.HTTPNotFound(explanation=e.format_mess

实际调用 get\_by\_id：

def get_aggregate(self, context, aggregate_id):
            """Get an aggregate by id."""
            return objects.Aggregate.get_by_id(context, aggregate_id)

get\_by\_id 依据id查询资源：

db_aggregate = _aggregate_get_from_db(context, aggregate_id)

@db_api.api_context_manager.reader
    def _aggregate_get_from_db(context, aggregate_id):
        query = context.session.query(api_models.Aggregate).\
                options(joinedload('_hosts')).\
                options(joinedload('_metadata'))
        query = query.filter(api_models.Aggregate.id == aggregate_id)
    
        aggregate = query.first()
    
        if not aggregate:
            raise exception.AggregateNotFound(aggregate_id=aggregate_id)
    
        return aggregate

该方法返回aggregates所有信息：  
包含三个表数据： aggregates， aggregate\_hosts， aggregate\_metadata

1.  更新：

(可以更新name)

@extensions.expected_errors((400, 404, 409))
        @validation.schema(aggregates.update_v20, '2.0', '2.0')
        @validation.schema(aggregates.update, '2.1')
        def update(self, req, id, body):
            """Updates the name and/or availability_zone of given aggregate."""
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'update')
            updates = body["aggregate"]
            if 'name' in updates:
                updates['name'] = common.normalize_name(updates['name'])
    
            try:
                aggregate = self.api.update_aggregate(context, id, updates)
            except exception.AggregateNameExists as e:
                raise exc.HTTPConflict(explanation=e.format_message())
            except exception.AggregateNotFound as e:
                raise exc.HTTPNotFound(explanation=e.format_message())
            except exception.InvalidAggregateAction as e:
                raise exc.HTTPBadRequest(explanation=e.format_message())
    
            return self._marshall_aggregate(req, aggregate)

self.api.update\_aggregate(context, id, updates)  
该步骤首先校验主机组id是否存在，然后更新数据进行保存  
`aggregate.name = values.pop('name')`可看出只能更新name  
可以更新az数据metadata  
更新完成后，刷新调度的数据

def update_aggregate(self, context, aggregate_id, values):
            """Update the properties of an aggregate."""
            aggregate = objects.Aggregate.get_by_id(context, aggregate_id)
            if 'name' in values:
                aggregate.name = values.pop('name')
                aggregate.save()
            self.is_safe_to_update_az(context, values, aggregate=aggregate,
                                      action_name=AGGREGATE_ACTION_UPDATE)
            if values:
                aggregate.update_metadata(values)
                aggregate.updated_at = timeutils.utcnow()
            self.scheduler_client.update_aggregates(context, [aggregate])
            .....

is\_safe\_to\_update\_az有如下处理，即是更新元数据

if 'availability_zone' in metadata:
                if not metadata['availability_zone']:
                    msg = _("Aggregate %s does not support empty named "
                            "availability zone") % aggregate.name
                    self._raise_invalid_aggregate_exc(action_name, aggregate.id,

删除，delete方法：

@extensions.expected_errors((400, 404))
        def delete(self, req, id):
            """Removes an aggregate by id."""
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'delete')
            try:
                self.api.delete_aggregate(context, id)
            except exception.AggregateNotFound as e:
                raise exc.HTTPNotFound(explanation=e.format_message())
            except exception.InvalidAggregateAction as e:
                raise exc.HTTPBadRequest(explanation=e.format_message())

delete\_aggregate首先查询主机组数据，校验存在性  
然后校验主机是否为空，为空的主机组才允许删除：

if len(aggregate.hosts) > 0:
                msg = _("Host aggregate is not empty")
                raise exception.InvalidAggregateActionDelete(
                    aggregate_id=aggregate_id, reason=msg)

然后调用aggregate.destroy() 删除数据

destroy：

def _aggregate_delete_from_db(context, aggregate_id):
        # Delete Metadata first
        context.session.query(api_models.AggregateMetadata).\
            filter_by(aggregate_id=aggregate_id).\
            delete()
    
        count = context.session.query(api_models.Aggregate).\
                    filter(api_models.Aggregate.id == aggregate_id).\
                    delete()
    
        if count == 0:
            raise exception.AggregateNotFound(aggregate_id=aggregate_id)

先删除metadata数据，然后清理aggregate表数据