Open stack的计量计费功能（一）

2023-05-16

简要

针对于一些云计算的产品来讲，实现资源的计量/计费功能是一个比较大的工程，很多公司都是基于OpenStack开发自己的计费服务作为其产品化的一部分。本文主要是针对OpenStack的计量计费根据自己的学习认知做一个总结

文中涉及到的一些链接：
cloudkitty运用
Ceilometer的认知

一如何实现

1.1 总述

完成资源的计费需要经过以下几个步骤：
在这里插入图片描述

步骤一：资源数据的收集工作，包含资源的使用对象（what），使用者（who），使用时间（when），使用量（how much）等基本信息，Ceilometer是专门为OpenStack环境提供一个获取和保存各种测量值的统一框架
步骤二：对【步骤一】的资源数据根据产品的计费策咯进行计费
步骤三：在【步骤一】【步骤二】已经完成资源的计费和存储的前提下，我们可以根据业务需求去合理的使用已经完成计费的数据，开发针对于业务层面的API

1.2 Ceilometer介绍

Ceilometer是专门为OpenStack环境提供一个获取和保存各种测量值的统一框架，明确目标，同时将其告警功能拆分到aodh项目，采集数据存储到gnocchi时间序列数据库，事件相关的服务拆分到panko项目，相关数据存储在MongoDB中

1.2.1 框架

在这里插入图片描述

Ceilometer：数据采集服务，采集资源使用量相关数据，并推送到Gnocchi，采集操作事件数据，并推送到Panko。
Gnocchi：时间序列数据库，保存计量数据。
Panko：事件数据库，保存事件数据。
Aodh：告警服务，基于计量和事件数据提供告警通知功能。

1.2.2 数据收集方式

数据收集方式从大方向上来看分为两种：主动/被动，下图为Ceilometer收集数据的方式，主要分为三大类：

通知【被动】：所有的OpenStack服务都会在执行了某种操作或者状态变化时发送通知消息到oslo-messaging（OpenStack整体的消息队列框架），一些消息中包含了计量所需要的数据，这部分消息会被Ceilometer的ceilometer-agent-notification服务组件处理，并转化为samples。通知数据采集方式是被动地采集计量数据。
轮询【主动】：Ceilometer中的服务组件根据配置定期主动地通过OpenStack服务的API或者其他辅助工具（如Hypervisors）去远端或本地的不同服务实体中获取所需要的计量数据；Ceilometer的轮询机制通过3种类型的代理实现，即ceilometer-agent-central、ceilometer-agent-compute和ceilometer-agent-ipmi服务组件。每种代理使用不同的轮询插件（pollster）从不同的命名空间来收集数据。
REST ful API【被动】：用户可以通过调用RESTful API直接把利用其他方式获取的任意测量数据送达给Ceilometer。

1.2.3 数据处理方式

Ceilometer获取到测量值数据后，会把它转化为符合某种标准格式的数据采样（Sample）通过内部总线发送给Notification agent。然后Notification Agent根据用户定义的Pipeline来对数据采样进行转换（Transform）和发布（Publish）。如果根据Pipeline的定义，这个数据采样（Sample）最后被发布给Collector的话，Collector会把这个数据采样保存在数据库中。Ceilometer的计量数据经过数据采集（agent）、数据处理（流水线数据转换及发布）、数据存储（Storage）。Gnocchi是一个多租户时间序列，计量和资源数据库。提供了REST API接口来创建和操作数据，用于超大规模计量数据的存储，同时向操作者和用户提供对度量和资源信息的访问。

1.3 CloudKitty 计费

1.3.1 框架

在这里插入图片描述
整体上Cloudkitty计费引擎以定时（CONF.collect.period）task的形式执行计费任务；每一轮计费任务之初要知道需要为哪些租户进行计费get_tenants()，再依次对每个租户的每一项服务进行计费；首先会通过collector模块从计量数据源中获得计费数据data；其次将数据交给计费模型根据定价规则完成费用计算；最后使用storage模块将费用数据持久化存储保存。

1.3.2 CloudKitty - orchestrator(数据处理)

    def run(self):
        LOG.debug('Started worker {}.'.format(self._worker_id))
        while True:
            self.tenants = self.fetcher.get_tenants()
            random.shuffle(self.tenants)
            LOG.info('[Worker: {w}] {n} tena nts loaded for fetcher {f}'.format(
                w=self._worker_id, n=len(self.tenants), f=self.fetcher.name))

            for tenant_id in self.tenants:

                lock_name, lock = get_lock(self.coord, tenant_id)
                LOG.debug(
                    '[Worker: {w}] Trying to acquire lock "{lck}" ...'.format(
                        w=self._worker_id, lck=lock_name)
                )
                if lock.acquire(blocking=False):
                    LOG.debug(
                        '[Worker: {w}] Acquired lock "{lck}" ...'.format(
                            w=self._worker_id, lck=lock_name)
                    )
                    state = self._check_state(tenant_id)
                    if state:
                        worker = Worker(
                            self.collector,
                            self.storage,
                            tenant_id,
                            self._worker_id,
                        )
                        worker.run()

                    lock.release()

            # FIXME(sheeprine): We may cause a drift here
            time.sleep(CONF.collect.period)

…\cloudkitty\orchestrator.py
以上代码是CloudKitty的run函数，从函数当中整个计费分为以下几个部分：

获取到租户信息
循环租户列表，检查计费情况
在Worker中进行数据收集，数据计量，数据存储操作

    def run(self):
        while True:
            # 判读是否到计费周期了，timestamp为起始周期，一般获取到的都是当前第一个计费周期，即该月的起止时间
            timestamp = self._check_state()
            if not timestamp:
                break

            #服务包括：compute,image,volume,network.bw.in,network.bw.out,network.floating
            metrics = list(self._conf['metrics'].keys())

            # Collection
            usage_data = self._do_collection(metrics, timestamp)

            frame = dataframe.DataFrame(
                start=timestamp,
                end=tzutils.add_delta(timestamp,
                                      timedelta(seconds=self._period)),
                usage=usage_data,
            )
            # Rating
            # 根据stevestore的插件功能，查询’cloudkitty.rating.processors’提供的计费策略，并进行计费，当前主要讲hashmap的计费代码
            for processor in self._processors:
                frame = processor.obj.process(frame)

            # Writing
            self._storage.push([frame], self._tenant_id)
            self._state.set_state(self._tenant_id, timestamp)

1.3.3 获取到租户的信息-Fetcher

在这里插入图片描述
目前支持五种获取租户的方式，除了PrometheusFetcher外每种方式里面都提供了获取租户的方法get_tenants
Openstack文档连接： https://docs.openstack.org/cloudkitty/latest/admin/configuration/fetcher.html

GnocchiFetcher
Gnocchi中存储了不同种类资源的数据
KeystoneFetcher
Keystone是默认的获取计费租户的方式，同时支持V2和V3版本。具体逻辑是检查Cloudkitty用户是否在某个tenant内并拥有rating角色。所以在使用Cloudkitty时，一般对于想要计费的租户需要执行命令’keystone user-role-add --user cloudkitty --role rating --tenant demo’，将Cloudkitty用户加入租户并赋予rating角色。
MonascaFetcher
Monasca：一个具有高性能，可扩展，高可用的监控即服务的（MONaas）解决方案。Monasca 是一个多租户监控即服务工具，可以帮助IT团队分析日志数据并设置告警和通知
PrometheusFetcher
Prometheus针对资源的监控收集到的数据，获取的是监控数据的对象作为租户
SourceFetcher
从配置文件中获取

1.3.4 获取数据-Controller

Open stack文档链接：https://docs.openstack.org/cloudkitty/latest/developer/collector.html
在这里插入图片描述
Controller的数据来源主要分为三个部分：

GnocchiCollector
MonascaCollector
PrometheusCollector
上面的三种收集方式中gnocchi为默认的数据收集方式，每种收集方式都会有一个fetch_all函数，该方法会收集到数据然后用在retrieve函数当中，最终返回统一的数据格式
fetch_all函数的数据收集是依赖于/etc/cloudkitty/metrics.yml文件的，该文件中对要统计的资源进行了配置，数据的收集以及处理都依赖该配置文件

metrics:
  cpu:
    unit: instance  # metrics的计量单位
    alt_name: instance  # 对该计量项起了一个别名
    groupby:  # 数据分类的依据
      - id
      - user_id
      - project_id
    metadata:  # 数据计量的依据
      - flavor_name
      - flavor_id
      - vcpus
    mutate: NUMBOOL # 转化的数据类型
    extra_args: # 计量时所依赖的一些参数
      aggregation_method: mean
      resource_type: instance
      force_granularity：60  #代表着多少秒
      re_aggregation_method：rate:
      re_aggregation_method：sum  # 以防检索到的聚合需要重新聚合
      use_all_resource_revisions：True # 该配置默认为True 用在过滤数据的方法中作为判断依据（filter_unecessary_measurements）

配置文件中的mutate类型

def mutate(value, mode='NONE'):
    """Mutate value according provided mode."""

    if mode == 'NUMBOOL':
        return float(value != 0.0)

    if mode == 'FLOOR':
        return math.floor(value)

    if mode == 'CEIL':
        return math.ceil(value)

    return value

下面分析下GnocchiCollector的数据收集方式

根据metric_name在配置文件中获取该metric的全部配置
获取时间范围内metric的数据，如果该metric的extra_args中存在force_granularity>0并且re_aggregation_method.startswith(‘rate:’)，则下面的方法会返回时间范围为【start - timedelta(seconds=force_granularity)】-【end】的数据
对2 获取到的数据进行处理，过滤掉不需要的数据，默认是返回全部
根据配置文件中的metadata获取资源的数据
进行数据的格式化处理

    def fetch_all(self, metric_name, start, end,
                  project_id=None, q_filter=None):
                  
		# 1 根据metric_name在配置文件中获取该metric的全部配置
        met = self.conf[metric_name]
		# 2 获取时间范围内metric的数据 ，如果该metric的extra_args中存在force_granularity>0并且re_aggregation_method.startswith('rate:')，则下面的方法会返回时间范围为【start - timedelta(seconds=force_granularity)】-【end】的数据
        data = self._fetch_metric(
            metric_name,
            start,
            end,
            project_id=project_id,
            q_filter=q_filter,
        )
		# 3 对2 获取到的数据进行处理，过滤掉不需要的数据，默认是返回全部
        data = GnocchiCollector.filter_unecessary_measurements(
            data, met, metric_name)
		
		# 4 根据配置文件中的metadata获取资源的数据
        resources_info = None
        if met['metadata']:
            resources_info = self._fetch_resources(
                metric_name,
                start,
                end,
                project_id=project_id,
                q_filter=q_filter
            )
		# 5 进行数据的格式化处理  
        formated_resources = list()
        for d in data:
            # Only if aggregates have been found
            LOG.debug("Processing entry [%s] for [%s] in timestamp ["
                      "start=%s, end=%s] and project id [%s]", d,
                      metric_name, start, end, project_id)
            if d['measures']['measures']['aggregated']:
                try:
                 # 在该方法中会进行单位的处理
                    metadata, groupby, qty = self._format_data(
                        met, d, resources_info)
                except AssociatedResourceNotFound as e:
                    LOG.warning(
                        '[{}] An error occured during data collection '
                        'between {} and {}: {}'.format(
                            project_id, start, end, e),
                    )
                    continue
                formated_resources.append(dataframe.DataPoint(
                    met['unit'],
                    qty,
                    0,
                    groupby,
                    metadata,
                ))
        return formated_resources

 @abc.abstractmethod
    def fetch_all(self, metric_name, start, end,
                  project_id=None, q_filter=None):
        ""
        Fetches information about a specific metric for a given period.

        This method must respect the ``groupby`` and ``metadata`` arguments
        provided in the metric conf at initialization.
        (Available in ``self.conf['groupby']`` and ``self.conf['metadata']``).

        Returns a list of cloudkitty.dataframe.DataPoint objects.

        :param metric_name: Name of the metric to fetch
        :type metric_name: str
        :param start: start of the period
        :type start: datetime.datetime
        :param end: end of the period
        :type end: datetime.datetime
        :param project_id: ID of the scope for which data should be collected
        :type project_id: str
        :param q_filter: Optional filters
        :type q_filter: dict
        ""

    def retrieve(self, metric_name, start, end,
                 project_id=None, q_filter=None):

        data = self.fetch_all(
            metric_name,
            start,
            end,
            project_id,
            q_filter=q_filter,
        )

        name = self.conf[metric_name].get('alt_name', metric_name)
        if not data:
            raise NoDataCollected(self.collector_name, name)

        return name, data

1.3.5 数据计量-Rating

Openstack地址：https://docs.openstack.org/cloudkitty/latest/user/rating/index.html
Cloudkitty当前的计费模型有三个，分别是noop，hashmap和pyscripts。

noop模型为空，仅作为测试用；
hashmap是当前Cloudkitty实用价值最高，最容易使用，最接近实际案例的计费模型
pyscripts计费模型提供了使用python代码定制计费的接口，用户可以直接将含有计费逻辑的python脚本上传给cloudkitty实现定制化的计费，所以pyscripts计费模型使用门槛较高。

Hashmap Mapping 方式

resource_type	Flat	rate	base_price
基础型	200	1.2	200*1.2=240
计算型	400	1.2	400*1.2=480
存储型	200	200	200

Hashmap Threshold 方式

resource_type	Flat	rate	level	base_price
高效云盘	20	0.9	20G	20*0.9=18
高效云盘	20	0.8	50G	20*0.8=16
ESSD云盘	40	0.8	50G	40*0.8=32

1.3.6 数据存储-Stroge

Openstack地址：https://docs.openstack.org/cloudkitty/latest/developer/storage.html
在这里插入图片描述
Stroge存在V和V2两个版本，open stack中只对V2版本进行了描述，并表示该功能是不稳定的

计费模型计算出来的费用数据将由storage模块持久化存储下来，所需记录的核心字段包括begin，end，unit，qty，res_type，desc和tenant_id等，包含了时间信息，资源相关信息，属主信息。每个租户的各项服务费用数据会先缓存下来，最后再将当前租户的当前周期内所产生的费用数据一次性提交到存储后端。

Storage的具体实现过程包括：
（1）通过抽象方法get_time_frame从存储后端中的数据来确定下一个时间范围。
（2）通过append方法将费用数据记入提交缓存，期间可能会通过get_tenants函数和_dispatch函数做预处理或加工。
（3）通过commit函数将费用数据写入到后端存储持久化。这个过程会做_pre_commit，_commit，_post_commit一系列的工作确保数据被可靠地存储。
（4）对外提供get_total方法，返回费用情况。 sqlalchemy和gnocchi_hybrid的数据格式或者说表的结构都包括begin, end, unit, qty, res_type, rate, tenant_id和对资源描述相关的字段。实际上两种后端存储都是基于SQL实现的，所以随着时间的推移和云环境中需要计费资源的增加，费用数据的增加同样会出现之前类似ceilometer用SQL存储measurement数据带来性能瓶颈的问题。所以将费用数据存储在gnocchi中是必须的，详见https://review.openstack.org/#/c/319425/。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)