openstack-nova源码分析(八）aggregates主机组管理 metadata

淡淡的烟草味﹌ 2022-02-19 07:17 330阅读 0赞

aggregate 主机管理管理  
可以向主机中添加，或者移除一个主机，也可以管理metadata

#### 一. URL 及命令行： ####

/os-aggregates/{aggregate_id}/action

命令行：

[root@nova ~]# nova help | grep agg
        aggregate-add-host          Add the host to the specified aggregate.
        aggregate-remove-host       Remove the specified host from the specified aggregate.
        aggregate-set-metadata      Update the metadata associated with the aggregate.

二. 源码分析：

aggregate 主机管理功能也是定义在AggregateController 中：  
class AggregateController(wsgi.Controller):

1.  添加主机操作：

@wsgi.action('add_host')
        @validation.schema(aggregates.add_host)
        def _add_host(self, req, id, body):
            """Adds a host to the specified aggregate."""
            host = body['add_host']['host']
    
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'add_host')
            try:
                aggregate = self.api.add_host_to_aggregate(context, id, host)
            except (exception.AggregateNotFound,
                    exception.ComputeHostNotFound) as e:
                raise exc.HTTPNotFound(explanation=e.format_message())
            except (exception.AggregateHostExists,
                    exception.InvalidAggregateAction) as e:
                raise exc.HTTPConflict(explanation=e.format_message())
            return self._marshall_aggregate(req, aggregate)

调用add\_host\_to\_aggregate  
定义如下：

def add_host_to_aggregate(self, context, aggregate_id, host_name):
            """Adds the host to an aggregate."""
            aggregate_payload = {'aggregate_id': aggregate_id,
                                 'host_name': host_name}
            compute_utils.notify_about_aggregate_update(context,
                                                        "addhost.start",
                                                        aggregate_payload)
            # validates the host; ComputeHostNotFound is raised if invalid
            objects.Service.get_by_compute_host(context, host_name)
    
            aggregate = objects.Aggregate.get_by_id(context, aggregate_id)
            self.is_safe_to_update_az(context, aggregate.metadata,
                                      hosts=[host_name], aggregate=aggregate)
    
            aggregate.add_host(host_name)
            self.scheduler_client.update_aggregates(context, [aggregate])
            self._update_az_cache_for_host(context, host_name, aggregate.metadata)
            # NOTE(jogo): Send message to host to support resource pools
            self.compute_rpcapi.add_aggregate_host(context,
                    aggregate=aggregate, host_param=host_name, host=host_name)
            aggregate_payload.update({'name': aggregate.name})
            compute_utils.notify_about_aggregate_update(context,
                                                        "addhost.end",
                                                        aggregate_payload)
            return aggregate

传入aggregate\_id, host\_name两个参数  
notify\_about\_aggregate\_update 写入日志  
get\_by\_compute\_host 获取主机信息，用于检查主机是否存在  
Aggregate.get\_by\_id 获取主机组信息  
aggregate.add\_host(host\_name)将主机添加到主机组中  
scheduler\_client.update\_aggregates 刷新调度  
\_update\_az\_cache\_for\_host 刷新az信息

aggregate.add\_host(host\_name)源码如下：

@base.remotable
        def add_host(self, host):
            if self.in_api:
                _host_add_to_db(self._context, self.id, host)
            else:
                db.aggregate_host_add(self._context, self.id, host)
    
            if self.hosts is None:
                self.hosts = []
            self.hosts.append(host)
            self.obj_reset_changes(fields=['hosts'])

最终使用api\_models.AggregateHost() update 将主机信息添加到aggregate\_hosts中

1.  remove host 删除主机  
    跟add host类似，这里只做简单分析:

@wsgi.action('remove_host')
       @validation.schema(aggregates.remove_host)
       def _remove_host(self, req, id, body):
           """Removes a host from the specified aggregate."""
           host = body['remove_host']['host']
    
           context = _get_context(req)
           context.can(aggr_policies.POLICY_ROOT % 'remove_host')
           try:
               aggregate = self.api.remove_host_from_aggregate(context, id, host)
           except (exception.AggregateNotFound, exception.AggregateHostNotFound,
                   exception.ComputeHostNotFound):
               msg = _('Cannot remove host %(host)s in aggregate %(id)s') % {
                           'host': host, 'id': id}
               raise exc.HTTPNotFound(explanation=msg)
           except exception.InvalidAggregateAction:
               msg = _('Cannot remove host %(host)s in aggregate %(id)s') % {
                           'host': host, 'id': id}
               raise exc.HTTPConflict(explanation=msg)
           return self._marshall_aggregate(req, aggregate)

实际调用remove\_host\_from\_aggregate 移除主机

remove\_host\_from\_aggregate步骤：  
notify\_about\_aggregate\_update写入日志  
get\_by\_compute\_host校验主机  
Aggregate.get\_by\_id获取主机组  
aggregate.delete\_host(host\_name)移除主机  
刷新scheduler\_client 调度，  
刷新az

aggregate.delete\_hos 源码：

@base.remotable
      def delete_host(self, host):
          if self.in_api:
              _host_delete_from_db(self._context, self.id, host)
          else:
              db.aggregate_host_delete(self._context, self.id, host)
    
          self.hosts.remove(host)
          self.obj_reset_changes(fields=['hosts'])

这里实际是调用api\_models.AggregateHost的delete方法进行删除

1.  metadata管理：  
    由于只有一个接口，并且只定义在一个动作中，因此metadata的管理，创建更新等操作都在一起，所以依据传入的数据为最终的数据，缺少的数据，则认为删除

源码如下：

@wsgi.action('set_metadata')
        @validation.schema(aggregates.set_metadata)
        def _set_metadata(self, req, id, body):
            """Replaces the aggregate's existing metadata with new metadata."""
            context = _get_context(req)
            context.can(aggr_policies.POLICY_ROOT % 'set_metadata')
    
            metadata = body["set_metadata"]["metadata"]
            try:
                aggregate = self.api.update_aggregate_metadata(context,
                                                               id, metadata)
            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)

实际调用： update\_aggregate\_metadata  
源码：

def update_aggregate_metadata(self, context, aggregate_id, metadata):
            """Updates the aggregate metadata."""
            aggregate = objects.Aggregate.get_by_id(context, aggregate_id)
            self.is_safe_to_update_az(context, metadata, aggregate=aggregate,
                                      action_name=AGGREGATE_ACTION_UPDATE_META)
            aggregate.update_metadata(metadata)
            self.scheduler_client.update_aggregates(context, [aggregate])
            # If updated metadata include availability_zones, then the cache
            # which stored availability_zones and host need to be reset
            if metadata and metadata.get('availability_zone'):
                availability_zones.reset_cache()
            aggregate.updated_at = timeutils.utcnow()
            return aggregate

更新az信息  
更新metadata  
更新调度信息  
刷新可用区cache

aggregate.update\_metadata(metadata)  
源码：

def update_metadata(self, updates):
            if self.in_api:
                metadata_delete = _metadata_delete_from_db
                metadata_add = _metadata_add_to_db
            else:
                metadata_delete = db.aggregate_metadata_delete
                metadata_add = db.aggregate_metadata_add
    
            payload = {'aggregate_id': self.id,
                       'meta_data': updates}
            compute_utils.notify_about_aggregate_update(self._context,
                                                        "updatemetadata.start",
                                                        payload)
            to_add = {}
            for key, value in updates.items():
                if value is None:
                    try:
                        metadata_delete(self._context, self.id, key)
                    except exception.AggregateMetadataNotFound:
                        pass
                    try:
                        self.metadata.pop(key)
                    except KeyError:
                        pass
                else:
                    to_add[key] = value
                    self.metadata[key] = value
            metadata_add(self._context, self.id, to_add)
            compute_utils.notify_about_aggregate_update(self._context,
                                                        "updatemetadata.end",
                                                        payload)
            self.obj_reset_changes(fields=['metadata'])

步骤：

notify\_about\_aggregate\_update 写入日志  
如果value值为空，则默认删除

metadata\_add检查需要更新的值，以及需要新增的

metadata_add(self._context, self.id, to_add)

校验更新或新增， 部分源码如下：

already_existing_keys = set()
                    if all_keys:
                        query = query.filter(
                            api_models.AggregateMetadata.key.in_(all_keys))
                        for meta_ref in query.all():
                            key = meta_ref.key
                            meta_ref.update({"value": metadata[key]})
                            already_existing_keys.add(key)
    
                    new_entries = []
                    for key, value in metadata.items():
                        if key in already_existing_keys:
                            continue
                        new_entries.append({"key": key,
                                            "value": value,
                                            "aggregate_id": aggregate_id})
                    if new_entries:
                        context.session.execute(
                            api_models.AggregateMetadata.__table__.insert(),
                            new_entries)