📈 第07章 · 监控报告脚本实战

Python + 飞书 Webhook 实践

监控报告脚本实战¶

这一章用一台真实服务器在跑的脚本当例子,讲"如何用 Python 把 Prometheus 数据取出来、画成图、推送到聊天群"。本机 /root/scripts/monitor_report.py 就是这么做的:从 Prometheus 取数 → 用 matplotlib 画多面板仪表盘 → 上传图片 → 发给飞书群,由 cron 每天 8:00 执行。

脚本需求拆解¶

环节	要做的事	用的工具
取数	从 Prometheus 拿时序数据	`urllib` + Prometheus HTTP API
出图	画环形图/趋势图/多线图	`matplotlib` + `numpy` + `pillow`
推送	发卡片/图片到群	`urllib` + 飞书 Webhook/开放接口
调度	每天定时跑	cron

这台机器上用 apt 装了 matplotlib、numpy、python3-pil,脚本用系统 python3 直接跑。

1) 从 Prometheus 取数¶

Prometheus 提供两个查询接口:

/api/v1/query 即时值(instant)
/api/v1/query_range 一段时间序列(range)

import urllib.request, urllib.parse, json

def query_range(server, query, start, end, step='60s'):
    p = urllib.parse.urlencode({
        'query': query, 'start': start, 'end': end, 'step': step})
    with urllib.request.urlopen(f'{server}/api/v1/query_range?{p}', timeout=10) as r:
        return json.loads(r.read())

def query_instant(server, query):
    p = urllib.parse.urlencode({'query': query})
    with urllib.request.urlopen(f'{server}/api/v1/query?{p}', timeout=10) as r:
        return json.loads(r.read())

关键 | Prometheus 返回结构:data.result[] 是序列数组,每个序列有 metric(标签)和 values(时间戳+值)。要自己抽 values 里的 [ts, value] 出来画坐标轴。

2) 用 matplotlib 画图¶

import matplotlib
matplotlib.use('Agg')          # 无显示环境必备,必须放 import pyplot 之前
import matplotlib.pyplot as plt
import io

def draw_ring(ax, val, title, unit, max_val=100):
    ax.clear()
    ax.pie([val, max(0, max_val - val)], colors=['#4ade80', '#1a1d29'],
           startangle=90, counterclock=False)
    ax.text(0, 0, f'{val:.1f}{unit}', ha='center', va='center', fontsize=18, color='white')
    ax.set_title(title)

然后用 plt.subplots 摆多面板,最后:

buf = io.BytesIO()
fig.savefig(buf, format='png', dpi=150, bbox_inches='tight')
buf.seek(0)

注意 | 想要 plt.show(),在无 GUI 的服务器上会报错,所以设置 matplotlib.use('Agg') 并改成 savefig 输出到文件或 BytesIO。

3) 中文字体问题¶

matplotlib 默认字体不含中文,标题会变方框。脚本里注册了常见中文字体路径:

import matplotlib.font_manager as fm
_FONT_PATHS = [
    '/usr/share/fonts/truetype/wqy/wqy-microhei.ttc',
    '/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc',   # 本机常见 Noto CJK
    'C:/Windows/Fonts/msyh.ttc',
]
for fp in _FONT_PATHS:
    if os.path.exists(fp):
        fm.fontManager.addfont(fp)
        plt.rcParams['font.sans-serif'] = [fm.FontProperties(fname=fp).get_name()]
        plt.rcParams['axes.unicode_minus'] = False   # 负号正常显示
        break

4) 推送到飞书¶

飞书机器人两种:

自定义 Webhook 机器人:换到一个 URL,{"msg_type":"text","content":{"text":"..."}}。简单。
应用机器人:用 app_id/app_secret 换 tenant_access_token,再调发送/上传接口。本机脚本走的是应用机器人(因为要传图片、发富文本卡片)。

def get_token(conf):
    body = json.dumps({'app_id': conf['app_id'], 'app_secret': conf['app_secret']}).encode()
    req = urllib.request.Request('https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
                                 data=body, headers={'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.loads(r.read())['tenant_access_token']

上传图片→拿 image_key:调用 open-apis/im/v1/images(multipart/form-data)。
发消息→调用 open-apis/im/v1/messages,带上 receive_id_type 与 content(含 image_key)。

def send_card(card):
    body = json.dumps({'msg_type': 'interactive', 'content': json.dumps(card)}).encode()
    req = urllib.request.Request(conf['webhook'] or IM_SEND_URL, data=body,
                                 headers={'Content-Type': 'application/json; charset=utf-8'})
    with urllib.request.urlopen(req, timeout=15) as r:
        return json.loads(r.read())

提示 | 配置统一放一个文件(本机 /root/.feishu_monitor.conf),脚本运行时读取。这样改群/改密钥不用动代码;文件要 chmod 600。

5) 用 cron 每天跑 + 日志¶

0 8 * * * /usr/bin/python3 /root/scripts/monitor_report.py >> /var/log/monitor_report.log 2>&1

健壮性要点(脚本里都应遵守):

每个外部请求设 timeout(timeout=10/15),避免卡死。
取数/画图/上传任一步出错要 try/except 记日志,别静默失败。
进程加锁:flock -n /tmp/mon_report.lock 防并发。
失败要能感知(或发告警),别做"哑脚本"。