Pythond 是定时触发用户自定义 Python 采集脚本的一整套方案。本文以 “获取每个小时登录的用户数”作为指标上报给中心为例。

1.1. 业务演示介绍

业务流程大致如下: 从数据库中采集数据 (Python 脚本) -> pythond 采集器定时触发该脚本上报数据(datakit) -> 从中心可看到指标(web)。

数据库现在有一张表叫 customers, 表中有如下字段:

  • name: 姓名 (字符串)
  • last_logined_time : 登录时间 (时间戳)

建表语句如下:

  1. create table customers
  2. (
  3.   `id`                BIGINT(20)  not null AUTO_INCREMENT COMMENT '自增 ID',
  4.   `last_logined_time` BIGINT(20)  not null DEFAULT 0      COMMENT '登录时间 (时间戳)',
  5.   `name`              VARCHAR(48) not null DEFAULT ''     COMMENT '姓名',
  7.   primary key(`id`),
  8.   key idx_last_logined_time(last_logined_time)
  9. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

往上面的表中插入测试数据:

  1. INSERT INTO customers (id, last_logined_time, name) VALUES (1, 1645600127, 'zhangsan');
  2. INSERT INTO customers (id, last_logined_time, name) VALUES (2, 1645600127, 'lisi');
  3. INSERT INTO customers (id, last_logined_time, name) VALUES (3, 1645600127, 'wangwu');

使用以下 SQL 语句来获取 “获取每个小时登录的用户数”:

  1. select count(1) from customers where last_logined_time>=(unix_timestamp()-3600);

把上面的数据以指标形式上报给中心。

下面详细讲述实现上述业务的具体步骤。

1.2. 前置条件

1.2.1. Python 环境

需要安装 Python，目前 Pythond 采集器处于 alpha 阶段，同时兼容 Python 2.7+ 和 Python 3+。但为了以后的兼容性，强烈建议使用 Python 3，毕竟 Python 2 官方已经不作支持了。下面的演示也是使用 Python 3。

1.2.2. Python 依赖库

需要安装以下依赖库:

  • requests（操作网络，用于上报指标）
  • pymysql（操作 MySQL 数据库，用于连接数据库获取业务数据）

安装方法如下:

  1. # python3
  2. python3 -m pip install requests
  3. python3 -m pip install pymysql

上述的安装需要安装 pip，如果你没有，可以参考以下方法(源自: 这里):

  1. # Linux/MacOS
  2. python3 -m ensurepip --upgrade
  4. # Windows
  5. py -m ensurepip --upgrade

1.3. 编写用户自定义脚本

需要用户继承 DataKitFramework 类，然后对 run 方法进行改写。DataKitFramework 类源代码文件是 datakit_framework.py，路径是 datakit/python.d/core/datakit_framework.py。

具体的使用可以参见源代码文件 datakit/python.d/core/demo.py。

我们这里根据上述需求，写成如下的 Python 脚本，命名为 hellopythond.py:

  1. from datakit_framework import DataKitFramework
  2. import pymysql
  3. import re
  4. import logging
  6. class MysqlConn():
  7.     def __init__(self, logger, config):
  8.         self.logger = logger
  9.         self.config = config
  10.         self.re_errno = re.compile(r'^\((\d+),')
  12.         try:
  13.             self.conn = pymysql.Connect(**self.config)
  14.             self.logger.info("pymysql.Connect() ok, {0}".format(id(self.conn)))
  15.         except Exception as e:
  16.             raise e
  18.     def __del__(self):
  19.         self.close()
  21.     def close(self):
  22.         if self.conn:
  23.             self.logger.info("conn.close() {0}".format(id(self.conn)))
  24.             self.conn.close()
  27.     def execute_query(self, sql_str, sql_params=(), first=True):
  28.         res_list = None
  29.         cur = None
  30.         try:
  31.             cur = self.conn.cursor()
  32.             cur.execute(sql_str, sql_params)
  33.             res_list = cur.fetchall()
  34.         except Exception as e:
  35.             err = str(e)
  36.             self.logger.error('execute_query: {0}'.format(err))
  37.             if first:
  38.                 retry = self._deal_with_network_exception(err)
  39.                 if retry:
  40.                     return self.execute_query(sql_str, sql_params, False)
  41.         finally:
  42.             if cur is not None:
  43.                 cur.close()
  44.         return res_list
  46.     def execute_write(self, sql_str, sql_params=(), first=True):
  47.         cur = None
  48.         n = None
  49.         err = None
  50.         try:
  51.             cur = self.conn.cursor()
  52.             n = cur.execute(sql_str, sql_params)
  53.         except Exception as e:
  54.             err = str(e)
  55.             self.logger.error('execute_query: {0}'.format(err))
  56.             if first:
  57.                 retry = self._deal_with_network_exception(err)
  58.                 if retry:
  59.                     return self.execute_write(sql_str, sql_params, False)
  60.         finally:
  61.             if cur is not None:
  62.                 cur.close()
  63.         return n, err
  65.     def _deal_with_network_exception(self, stre):
  66.         errno_str = self._get_errorno_str(stre)
  67.         if errno_str != '2006' and errno_str != '2013' and errno_str != '0':
  68.             return False
  69.         try:
  70.             self.conn.ping()
  71.         except Exception as e:
  72.             return False
  73.         return True
  75.     def _get_errorno_str(self, stre):
  76.         searchObj = self.re_errno.search(stre)
  77.         if searchObj:
  78.             errno_str = searchObj.group(1)
  79.         else:
  80.             errno_str = '-1'
  81.         return errno_str
  83.     def _is_duplicated(self, stre):
  84.         errno_str = self._get_errorno_str(stre)
  85.         # 1062：字段值重复，入库失败
  86.         # 1169：字段值重复，更新记录失败
  87.         if errno_str == "1062" or errno_str == "1169":
  88.             return True
  89.         return False
  91. class HelloPythond(DataKitFramework):
  92.     __name = 'HelloPythond'
  93.     interval = 10 # 每 10 秒钟采集上报一次。这个根据实际业务进行调节，这里仅作演示。
  95.     # if your datakit ip is 127.0.0.1 and port is 9529, you won't need use this,
  96.     # just comment it.
  97.     # def __init__(self, **kwargs):
  98.     #     super().__init__(ip = '127.0.0.1', port = 9529)
  100.     def run(self):
  101.         config = {
  102.             "host": "172.16.2.203",
  103.             "port": 30080,
  104.             "user": "root",
  105.             "password": "Kx2ADer7",
  106.             "db": "df_core",
  107.             "autocommit": True,
  108.             # "cursorclass": pymysql.cursors.DictCursor,
  109.             "charset": "utf8mb4"
  110.         }
  112.         mysql_conn = MysqlConn(logging.getLogger(''), config)
  113.         query_str = "select count(1) from customers where last_logined_time>=(unix_timestamp()-%s)"
  114.         sql_params = ('3600')
  115.         n = mysql_conn.execute_query(query_str, sql_params)
  117.         data = [
  118.           {
  119.               "measurement": "hour_logined_customers_count", # 指标名称。
  120.               "tags": {
  121.                 "tag_name": "tag_value", # 自定义 tag，根据自己想要标记的填写，我这里是随便写的
  122.               },
  123.               "fields": {
  124.                 "count": n[0][0], # 指标，这里是每个小时登录的用户数
  125.               },
  126.           },
  127.         ]
  129.         in_data = {
  130.             'M':data,
  131.             'input': "pyfromgit"
  132.         }
  134.         return self.report(in_data) # you must call self.report here

1.4. 将自定义脚本放入正确的位置

在 Datakit 安装目录的 python.d 目录下新建一个文件夹，并命名为 hellopythond，这个文件夹名称要与上面编写的类名相同，即为 hellopythond。

然后将上面写好的脚本 hellopythond.py 放入此文件夹下，即最后的目录结构如下:

  1. ├── ...
  2. ├── datakit
  3. └── python.d
  4.     ├── core
  5.     │   ├── datakit_framework.py
  6.     │   └── demo.py
  7.     └── hellopythond
  8.         └── hellopythond.py

上面的 core 文件夹是 Pythond 的核心文件夹，不要动。

上面是在没有开启 gitrepos 功能的情况下，如果是开启了 gitrepos 功能，那么路径结构就是这样的:

  1. ├── ...
  2. ├── datakit
  3. ├── python.d
  4. ├── gitrepos
  5. │   └── yourproject
  6. │       ├── conf.d
  7. │       ├── pipeline
  8. │       └── python.d
  9. │           └── hellopythond
  10. │               └── hellopythond.py

1.5. 开启 pythond 配置文件

将 Pythond 配置文件复制出来。在 conf.d/pythond 目录下复制 pythond.conf.sample 为 pythond.conf，然后将配置成如下形式:

  1. [[inputs.pythond]]
  3.     # Python 采集器名称
  4.     name = 'some-python-inputs'  # required
  6.     # 运行 Python 采集器所需的环境变量
  7.     #envs = ['LD_LIBRARY_PATH=/path/to/lib:$LD_LIBRARY_PATH',]
  9.     # Python 采集器可执行程序路径(尽可能写绝对路径)
  10.     cmd = "python3" # required. python3 is recommended.
  12.     # 用户脚本的相对路径(填写文件夹，填好后该文件夹下一级目录的模块和 py 文件都将得到应用)
  13.     dirs = ["hellopythond"] # 这里填的是文件夹名，即类名

1.6. 重启 Datakit

  1. sudo datakit --restart

1.7. 效果图

如果一切顺利的话，大概 1 分钟内我们就能在中心看到指标图。
14.pythond.png

1.8. 参考文档