prometheus-openstack-exporter源码解析

发表于 |

简介

服务的功能

获取openstack集群中cinder,nova,neutron, swift,swift-account-usage中的关于虚机，云盘，网络，存储相关的数据整理发送给prometheus server

代码地址

github链接: https://github.com/CanonicalLtd/prometheus-openstack-exporter.git

分支: master(和tag 0.1.4的版本一致)

它是怎么工作的

  1. prometheus server端周期性来exporter端抓取数据 (这里的周期是在prometheus server中用scrape_interval定义的)
  2. 我们的exporter服务启动后，会开启port的监听，一旦监听到有请求，会调用exporter里的一些列函数，将整理好的metrics返回给prometheus server，数据采集成功

源码

在看源码之前，或者之中需要知道的prometheus_client的一些概念

关于prometheus_client需要知道的这些

registry：就是一个监控项仓库，可以有很多个collector
collector：就是一个自定义的监控项收集器，一个collector中可以有很多个metrics
metrics: 监控项
metrics类型：count，gauge，summary，histogram

  • count类型：计数器 特点：只增不减，可以是浮点型，不一定非要是整数
  • gauge类型：瞬时值 特点：随意增减，浮点整数
  • summary类型：存储了 quantile数据，针对于长尾效应设计的，可画出准确的正态分布图
  • histogram类型： 收集正态分布数据，通过划分bucket，但是只是计数，比较粗略，客户端性能开销跟count和gauge差不多，图的准确性不如summary高，可画出粗略一点的正态分布图

其中 gauge、histogram、summary这三种metrics都没在这个exporter中出现，为了metrics类型完整（一家人就是要整整齐齐），所以把这三个也贴出来了；
写一个简单的collector举例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from prometheus_client import Counter, Gauge, Histogram, Summary, \
    CollectorRegistry, generate_latest, start_http_server
import time
import random

class TestCollector():
    def __init__(self):
        self.registry = CollectorRegistry()
        self.count_metric = Counter('counter_metric_name',
                         'A counter metirc decription test')
        self.gauge_metric = Gauge('gauge_metric_name',
                                  'A gauge metric description test')
        self.histogram_metric = Histogram('histogram_metric_name',
                           'A histogram description test',
                           buckets=(-5, 0, 5))
        self.summary_metric = Summary('summary_metric_name',
                         'A summary description test',
                        ['label1', 'label2'])

    def test(self):
        self.count_metric.inc(1)
        self.gauge_metric.set(random.random())
        self.histogram_metric.observe(random.randint(-10, 10))
        self.summary_metric.labels('a', 'b').observe(17)

        return generate_latest(registry=self.registry)


if __name__ == '__main__':

    collector = TestCollector()
    start_http_server(8000)
    while True:
        collector.test()
        time.sleep(2)

获取到的metric如下，我省略掉了一些prometheus_client自带的变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# HELP summary_metric_name A summary description test
# TYPE summary_metric_name summary
summary_metric_name_count{label1="a",label2="b"} 1678.0
summary_metric_name_sum{label1="a",label2="b"} 28526.0
# TYPE summary_metric_name_created gauge
summary_metric_name_created{label1="a",label2="b"} 1.584496763390804e+09
# HELP gauge_metric_name A gauge metric description test
# TYPE gauge_metric_name gauge
gauge_metric_name 0.8280131429442902
# HELP counter_metric_name_total A counter metirc decription test
# TYPE counter_metric_name_total counter
counter_metric_name_total 1678.0
# TYPE counter_metric_name_created gauge
counter_metric_name_created 1.584496762311975e+09
# HELP histogram_metric_name A histogram description test
# TYPE histogram_metric_name histogram
histogram_metric_name_bucket{le="-5.0"} 457.0
histogram_metric_name_bucket{le="0.0"} 848.0
histogram_metric_name_bucket{le="5.0"} 1262.0
histogram_metric_name_bucket{le="+Inf"} 1678.0
histogram_metric_name_count 1678.0
histogram_metric_name_sum 327.0
# TYPE histogram_metric_name_created gauge
histogram_metric_name_created 1.584496762312121e+09

代码层逻辑实现

好了，现在开始干正事，代码的逻辑是下边这样的：

  1. 开启一个线程，周期性(这里的周期是配置文件中的cache_refresh_interval定义的)的在线程内
    1. 通过环境变量获取到openstack的client()
    2. 使用openstack的组件client()获取我们想要的数据，并将数据缓存至本地的二进制文件中
  2. 开启一个事件监听循环：
    1. 当有请求 /metrics的时候，在/metrics路由下 将各个组件的collector监控数据组合到一起，返回给客户端
    2. 各个组件(nova,cinder,neutron,swift)的collector类从上边已经缓存的缓存文件中纷纷读取自己的数据，做处理，放到自己的registry里，再返回到metrics路由

从程序入口开始

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
if __name__ == '__main__':
    parser = argparse.ArgumentParser(usage=__doc__,
                                     description='Prometheus OpenStack exporter',
                                     formatter_class=argparse.RawTextHelpFormatter)
    parser.add_argument('config_file', nargs='?',
                        help='Configuration file path',
                        default='/etc/prometheus/prometheus-openstack-exporter.yaml',
                        type=argparse.FileType('r'))
    args = parser.parse_args()

    # 设置log的level，然后把log写到本地，或将日志输出到流 sys.stderr，sys.stdout
    log.setLevel(logging.DEBUG)
    for logsock in ('/dev/log', '/var/run/syslog'):
        if path.exists(logsock):
            log.addHandler(logging.handlers.SysLogHandler(address=logsock))
    else:
        log.addHandler(logging.StreamHandler())
    # 加载下配置文件
    config = yaml.safe_load(args.config_file.read())
    numeric_log_level = getattr(logging, config.get('log_level', 'INFO').upper(), None)
    if not isinstance(numeric_log_level, int):
        raise ValueError('Invalid log level: %s' % config.get('log_level'))
    log.setLevel(numeric_log_level)

    # 开启获取集群数据的线程
    data_gatherer = None
    if data_gatherer_needed(config):
        data_gatherer = DataGatherer()
        data_gatherer.start()

    #  开启HTTPServer监听    
    server = ForkingHTTPServer(('', config.get('listen_port')), handler)
    server.serve_forever()

1.先加载到配置文件，2.初始化记录log，3.开启数据收集（data_gatherer.start()），4.启动事件监听（server.serve_forever()）
接下来先从数据收集开始

开启线程收集数据 | DataGatherer(Thread)

单独开启一个线程，周期性通过client端获取我们所想要的数据，重写Thread的run()方法来完成我们想要在线程中进行的操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class DataGatherer(Thread):
    def run():
        prodstack = {}
        while True:
            # 假如 prodstack = {} 写下这，每次刷新数据的时候都会把他先置空，再复制，倒不如直接update一步到位
            try:
                # 在这里获取client, 并将各个client的处理好的数据，更新到prodstack里
                keystone, nova, neutron, cinder = get_clients()
                prodstack.update(self._get_keystone_info(keystone))  # 这些client会单独处理自己的数据
               prodstack.update(self._get_neutron_info(neutron))
               prodstack.update(self._get_nova_info(nova, cinder, prodstack))
            except Exception:
                # Ignore failures, we will try again after refresh_interval.
                # Most of them are termporary ie. connectivity problmes
                # To alert on stale cache use openstack_exporter_cache_age_seconds metric
                log.critical("Error getting stats: {}".format(traceback.format_exc()))
            else:
                # 在本地的二进制文件中写入加密的获取到的client的数据，作为缓存
                with open(self.cache_file + '.new', "wb+") as f:
                    pickle.dump((prodstack, ), f, pickle.HIGHEST_PROTOCOL)
                rename(self.cache_file + '.new', self.cache_file)
                log.debug("Done dumping stats to {}".format(self.cache_file))
            self.duration = time() - start_time
            sleep(self.refresh_interval)   #间隔多久再重新获取一次数据

Q: 为什么要加缓存呢，实时获取数据不好吗？
A：实时获取数据对于监控来说当然更精确，但是因为我们的查询比较繁琐，在集群规模比较大的情况下，查询一次可能需要几分钟，频繁的调API去抓取或者多个exporter抓取集群数据，也可能会导致集群响应延迟更高，还会给集群造成负面影响。

HTTPServer的框架 | ForkingHTTPServer(..)

单独把起HTTPServer的框架摘（二声）出来， 起一个异步进程HTTPserver并监听， 相关的代码是这样的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class OpenstackExporterHandler(BaseHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        BaseHTTPRequestHandler.__init__(self, *args, **kwargs)

    def do_GET(self):
        ......
  
              
class ForkingHTTPServer(ForkingMixIn, HTTPServer):
    pass


def handler(*args, **kwargs):    
    OpenstackExporterHandler(*args, **kwargs)

if __name__ == '__main__':
    .......
    server = ForkingHTTPServer(('', config.get('listen_port')), handler)
    server.serve_forever()

自己创建的ForkingHTTPServer：
ForkingHTTPServer：这里的ForkingHTTPServer继承自HTTPServer和ForkingMixIn，主要作用，异步起个进程，监听socket；

引用的模块中的两个类，这俩类的功能：
HTTPServer：继承自SocketServer.TCPServer, HTTPServer主要实现启动事件监听和address port的绑定，并将请求交于handler处理

ForkingMIxIn是继承于SocketServer用来做异步处理的类，它重载了process_request()方法，当出现新请求时创建一个新的进程，把具体工作放到新的进程中执行.

再来看一下我们的处理请求的Handler类
OpenstackExporterHandler：继承自BaseHTTPRequestHandler
OpenstackExporterHandler中定义do_GET来处理请求

BaseHTTPRequestHandler 是http handler的基类，无法直接使用，需要定义请求处理函数，所以我们重写do_GET()函数处理请求

关于server和handler的关系

python http框架主要有server和handler组成，server主要是用于建立网络模型，例如利用epoll监听socket；handler用于处理各个就绪的socket，就是server是用来启动事件监听，handler是来处理事件的。

总结一下就是当执行到下面这段代码时：

1
2
server = ForkingHTTPServer(('', config.get('listen_port')), handler)
server.serve_forever()

开启一个事件循环监听，(如果address是本地的话，可以在本地看到绑定的address和port)，可以在本地看到并将开启一个新进程来处理接受到的请求;

获取client | get_clients()

接着看在线程中是怎么获取client()的
从当前环境变量中获取需要的认证信息，然后获取各个openstack组件的client()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_clients():
    ks_version = int(env.get('OS_IDENTITY_API_VERSION', 2))
    if ks_version == 2:
        from keystoneclient.v2_0 import client as keystone_client
        # Legacy v2 env vars 在v2版本里用到了这些环境变量
        # OS_USERNAME OS_PASSWORD OS_TENANT_NAME OS_AUTH_URL OS_REGION_NAME   
        ks_creds = get_creds_dict("username", "password", "tenant_name",  
                          "auth_url", "region_name")
        # ks_creds打印：{'username': 'admin', 'tenant_name': 'admin', 'password': 'admin', 'auth_url': 'http://192.168.1.106/identity/v3', 'region_name': 'RegionOne'}                  
        cacert = maybe_get_cacert()
        if cacert:
            ks_creds["cacert"] = cacert
        nova_creds = [2] + get_creds_list("username", "password", "tenant_name",
                                          "auth_url")
        cinder_creds = get_creds_list("username", "password", "tenant_name",
                                      "auth_url")
        # nova_creds打印：[2, 'admin', 'admin', 'admin', 'http://192.168.1.106/identity/v3']
        # cinder_creds打印：['admin', 'admin', 'admin', 'http://192.168.1.106/identity/v3']                              
        keystone = keystone_client.Client(**ks_creds)
        nova = nova_client.Client(*nova_creds, cacert=cacert)
        neutron = neutron_client.Client(**ks_creds)
        cinder = cinder_client.Client(*cinder_creds, cacert=cacert)
    elif ks_version == 3:
        from keystoneauth1.identity import v3
        from keystoneauth1 import session
        from keystoneclient.v3 import client
        # 在v3版本里用到了这些环境变量
        #  OS_USERNAME OS_PASSWORD OS_USER_DOMAIN_NAME OS_AUTH_URL      
        #  OS_PROJECT_DOMAIN_NAME OS_PROJECT_DOMAIN_ID OS_PROJECT_ID OS_DOMAIN_NAME
        ks_creds_domain = get_creds_dict(
            "username", "password", "user_domain_name", "auth_url",
            "project_domain_name", "project_name", "project_domain_id", "project_id")

        #ks_creds_domain输出：{'username': 'admin', 'user_domain_name': 'default', 'password': 'admin', 'auth_url': 'http://192.168.1.106/identity/v3'}    
        # Need non-domain creds to get full catalog

        ks_creds_admin = get_creds_dict(
            "username", "password", "user_domain_name", "auth_url",
            "project_domain_name", "project_name", "project_domain_id", "project_id")
        # ks_creds_admin输出： {'username': 'admin', 'user_domain_name': 'default', 'password': 'admin', 'auth_url': 'http://192.168.1.106/identity/v3'}    

        auth_domain = v3.Password(**ks_creds_domain)
        auth_admin = v3.Password(**ks_creds_admin)
        # Need to pass in cacert separately
        verify = maybe_get_cacert()
        if verify is None:
            verify = True
        sess_domain = session.Session(auth=auth_domain, verify=verify)
        sess_admin = session.Session(auth=auth_admin, verify=verify)

        keystone = client.Client(session=sess_domain)
        nova = nova_client.Client(2, session=sess_admin)
        neutron = neutron_client.Client(session=sess_admin)
        cinder = cinder_client.Client(session=sess_admin)

一个openrc示例：

1
2
3
4
5
6
7
8
9
10
export OS_USERNAME=admin
export OS_TENANT_NAME=admin
export OS_PASSWORD=admin
export OS_REGION_NAME=RegionOne
export OS_AUTH_URL=http://192.168.1.106/identity/v3
export OS_IDENTITY_API_VERSION=3
export OS_USER_DOMAIN_ID=default
export OS_USER_DOMAIN_NAME=default
export OS_PROJECT_DOMAIN_ID=default
export OS_PROJECT_NAME=admin

Handler返回数据 |OpenstackExporterHandler(..)

在OpenstackExporterHandler中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# 在这把collector的和具体实现collector的类的映射关系放到了字典里
COLLECTORS = {
    'cinder': Cinder,
    'neutron': Neutron,
    'nova': Nova,
    'swift': Swift,
    'swift-account-usage': SwiftAccountUsage,
    }

DATA_GATHERER_USERS = [
    'cinder',
    'neutron',
    'nova',
    ]
    
    
class OpenstackExporterHandler(BaseHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
        
    def do_GET(self):
        url = urlparse.urlparse(self.path)
        if url.path == '/metrics':
            try:
                # collectors取配置文件enable_collectors和本地可获取的所有collector的交集，
                collectors = [COLLECTORS[collector]() for collector in get_collectors(config.get('enabled_collectors'))]
                log.debug("Collecting stats..")
                output = ''
                # 通过collector的get_stats()获取单个collector整理好的自己registry
                for collector in collectors:
                output += collector.get_stats()
                # 
                if data_gatherer:
                    output += data_gatherer.get_stats()

                self.send_response(200)
                self.send_header('Content-Type', CONTENT_TYPE_LATEST)
                self.end_headers()
                self.wfile.write(output)
            except Exception:
                self.send_response(500)
                self.end_headers()
                self.wfile.write(traceback.format_exc())

OpenstackExporterHandler集成自BaseHTTPRequestHandler，通过重写do_GET()函数，响应请求

self.wfile也是BaseHTTPRequestHandler里方法，用来写入应答信息

self.send_header(..), self.send_response(..) 这些都是BaseHTTPRequestHandler中的方法

Collector怎么收集数据的 | Nova()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class Nova():
    def __init__(self):
        # 创建一个Collector仓库
        self.registry = CollectorRegistry()      
        self.prodstack = {}
        # 从我们之前创建的cache_file中读取数据
        with open(config['cache_file'], 'rb') as f:    
            self.prodstack = pickle.load(f)[0]
        self.hypervisors = self.prodstack['hypervisors']
            .......
            
    def gen_hypervisor_stats(self):
        # 注册监控项
        vms = Gauge('hypervisor_running_vms', 'Number of running VMs', labels, registry=self.registry)
        vcpus_total = Gauge('hypervisor_vcpus_total', 'Total number of vCPUs', labels, registry=self.registry)
        ......
        for h in self.hypervisors:
            # 给监控项赋值
            vms.labels(*label_values).set(squashnone(h['running_vms']))
            vcpus_total.labels(*label_values).set(squashnone(h['vcpus']))
        .....
  
    def get_stats(self):
        log.debug("get_stats")
        # 下面这些函数都是从缓存文件中获取数据，监控项注册并赋值
        self.gen_hypervisor_stats()
        self.gen_instance_stats()
        self.gen_overcommit_stats()
        self.gen_quota_stats()
        # 将收集到的监控项都注册到自己仓库中，并返回
        return generate_latest(self.registry)

通过get_stats()返回registry
再来看一下DataGatherer的get_stats()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class DataGatherer(Thread):
    def get_stats(self):
        registry = CollectorRegistry()
        labels = ['cloud']
        # age 是最新的缓存文件存在的时间
        age = Gauge('openstack_exporter_cache_age_seconds',
                    'Cache age in seconds. It can reset more frequently '
                    'than scraping interval so we use Gauge',
                    labels, registry=registry)
        label_values = [config['cloud']]
        age.labels(*label_values).set(time() - path.getmtime(self.cache_file))
        
        # duration 是重新获取一次数据client开始到返回结果的时间
        duration = Gauge('openstack_exporter_cache_refresh_duration_seconds',
                         'Cache refresh duration in seconds.',
                         labels, registry=registry)
        duration.labels(*label_values).set(self.duration)
        return generate_latest(registry)

metrics列表

待补充

grafana-dashboard

对应的dashboard都写好了：

https://grafana.com/grafana/dashboards/7924/reviews
无奈我没有swift，所以不show图了；

监控swift那块好像还是有点精彩的，我没有swift环境，
参考链接：

关于HTTPServer的：http://luodw.cc/2016/11/05/python-http/

metrics参考: https://segmentfault.com/a/1190000018372390

metrics参考2: https://www.infoq.cn/article/Prometheus-theory-source-code

源码解析：https://blog.csdn.net/qingyuanluofeng/article/details/84779737

显示 Gitment 评论