CloddsBot：基于OpenAPI与规则引擎的云资源自动化巡检实践

1. 为什么会有 CloddsBot：从“看板疲劳”到自动化值守

先交代一个背景：我手上同时维护着几个不同云厂商的账号，加起来有几十台云主机、数据库实例、负载均衡、对象存储桶。每天早上开工第一件事，就是挨个登录控制台，刷新实例列表，看看有没有人半夜把磁盘打满，或者证书快过期了。这种巡检式工作做一两个星期没什么，做一两个月就非常折磨——你明知道这些事情大概率没问题，但你又不敢不看。

CloddsBot 就是在这种情况下写出来的一个轻量巡检机器人。名字拆开看是 Cloud + Logs + Bot，思路很直接：它替我们把“盯云资源”这件事自动化掉。项目做的事情归结起来就三件：定时调用云厂商的 OpenAPI 拉取资源状态；用一组可配置规则判断哪些状态算“异常”；把异常信息推送到钉钉或飞书群，顺便留好操作入口。整套东西跑在一台 1C2G 的轻量云主机上，成本可以忽略。

这个项目适合谁？一种是账号多、实例多但又不愿意上一整套 Prometheus + Grafana 的个人开发者或小团队；另一种是想接触云厂商 OpenAPI 封装、定时任务调度、IM Webhook 集成，想看一个完整可落地例子的后端学习者。这篇文章我会把架构选型、核心代码逻辑、部署方式和踩过的坑都展开讲，代码量不大，复制下来改改配置就能用。

1.1 每天花半小时做重复劳动，问题不只是累

先说痛点。我最早也试过“云厂商自带监控 + 短信告警”，但实际用下来有几个别扭的地方：多账号之间切换要重新登录，不同厂商控制台设计不一致，默认告警规则又太粗——比如磁盘使用率超过 80% 就短信轰炸，而很多时候只是测试机的临时数据增长，并不值得半夜爬起来处理。真正需要关注的“实例意外宕机”“云盘即将到期”“SSL 证书链断裂”这类事件，反而没有现成规则能覆盖。

更麻烦的是，账号一多，控制台里的状态信息就变成了碎片。实例列表要看一遍，RDS 要看一遍，负载均衡要看一遍，对象存储桶的权限到期又要看一遍。看完还容易忘，第二天再重复一轮。时间一长就会意识到：这种巡检本质上是在做“状态比对”和“异常判断”，而这恰恰是程序最擅长的事情。

1.2 现有开源监控体系为什么“用不动”

我也认真评估过 Prometheus + Grafana + Alertmanager 这套标准组合。好处是生态成熟、指标能力强，但问题也明显：部署和升级有维护成本，告警规则要用 PromQL 写，学习曲线对于只想“盯几台机器”的场景来说有点重。如果目标是从云厂商 API 层面管资源生命周期，而不是从操作系统层面采指标，那这套体系的收益其实是打折的。

CloddsBot 选择的是另一条路：不采集 CPU 内存这些性能指标（这类需求用云监控就够了），只聚焦“状态类事件”——实例是否 running、到期时间是否临近、账单是否异常、日志里有没有新增 ERROR。状态类事件数量少、判定逻辑直白，用一个轻量规则引擎就能搞定，完全不需要庞大的采集器和时序数据库。

1.3 项目的三条设计原则

CloddsBot 从第一天起给自己定了三条规矩：第一，配置驱动，所有账号、区域、规则都写在 YAML 里，加账号不用改代码；第二，只读优先，默认只调用云厂商的查询类接口，涉及重启、扩容的操作全部走独立命令并记录审计；第三，推送必须有人看，告警进 IM 群比进邮箱有效得多，群里的历史记录还天然可检索。

2. 三个核心模块与选型逻辑：采集器、规则引擎、通知器

整个 CloddsBot 可以拆成三个模块，这也是大多数类似工具的标准拆法：采集器（Collector）负责对接各云厂商 OpenAPI；规则引擎（Evaluator）负责把采集到的原始状态翻译成“正常、异常、需要关注”；通知器（Notifier）负责把判定结果推送到 IM，并对接人工操作。三个模块之间通过队列或共享数据库解耦，现阶段数据量不大，直接用轮询 + SQLite 就够了。

2.1 数据流：定时触发、状态归并、结果落库

流程大概是这样的：APScheduler 按 cron 表达式触发一次巡检任务；采集器依次遍历配置中的每个账号和区域，拉取资源清单；规则引擎将原始字段归一到统一状态模型然后逐条匹配规则；命中异常的事件先写入 SQLite 的 events 表，再经过告警抑制判断，决定是否真正推送到群。所有历史事件落库还有一个好处：隔周想复盘“上周到底出了几次问题”，一条 SQL 就能查出来。

SELECT date(created_at), level, count(*) FROM events WHERE created_at >= datetime('now', '-7 days') GROUP BY date(created_at), level;

2.2 为什么是 Python + APScheduler 而不是 Crontab + Shell

选择 Python 几乎是必然的：主流云厂商都提供官方 SDK，pip install 就能用；异常处理和 JSON 解析写起来比 Shell 舒服太多。调度部分我没有用系统 Crontab，而是用 APScheduler 的 CronTrigger，原因是它支持错过任务补偿——如果进程在巡检时间点恰好重启，APScheduler 会在恢复后补跑，而 Crontab 错过了就是错过了。常驻进程模式也比每分钟唤起一个脚本来得更稳，依赖连接和日志句柄都能提前初始化。

调度配置示例：

from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job( run_instance_check, CronTrigger.from_crontab("*/5 * * * *", timezone="Asia/Shanghai"), id="instance_check", misfire_grace_time=60, ) scheduler.add_job( run_cert_check, CronTrigger.from_crontab("0 9 * * *", timezone="Asia/Shanghai"), id="cert_check", misfire_grace_time=3600, )

这里 misfire_grace_time 也是容易被忽略的参数：它表示任务原定执行时间点过后多少秒内仍可以补跑。设得太短，进程重启稍微慢一点就会跳过巡检；设得太长，低频任务可能在错误的时间点突然补跑一堆。我习惯状态检查给 60 秒，每天一次的检查给 1 小时。

2.3 通知通道：为什么选 IM 自定义机器人 Webhook

钉钉、飞书、企业微信都支持自定义机器人 Webhook，使用门槛最低，只需要出站网络请求，不需要暴露任何入站端口。这对安全很友好：CloddsBot 所在的主机可以不开任何公网入站规则。需要强调一点：这类 Webhook 是单向的，机器人只能往群里发消息，收不到群成员回复。所以我把“确认后执行操作”这个闭环放在了一个独立命令行工具里，而不是试图在群里做交互。这样设计是权衡过的：自定义机器人做交互往往要升级为企业自建应用，涉及权限申请和回调服务，对个人项目来说太重了。

2.4 配置即代码：YAML 与环境变量分家的做法

账号信息属于敏感内容，绝对不能写死在 YAML 里提交到 Git。我采用的模式是：YAML 只保存账号名、区域、规则这些非敏感内容；密钥通过环境变量或 docker-compose 的 env_file 注入，程序读取环境变量后组装 SDK 客户端。这样即使代码仓库泄露，攻击者拿到的也只是一串标识，拿不到真正的密钥。

# config/accounts.yaml accounts: - name: demo provider: aliyun regions: - cn-hangzhou - cn-beijing access_key_env: DEMO_ALIYUN_AK secret_key_env: DEMO_ALIYUN_SK - name: staging provider: tencent regions: - ap-guangzhou access_key_env: STAGING_TENCENT_AK secret_key_env: STAGING_TENCENT_SK

密钥在 .env 文件里管理：

DEMO_ALIYUN_AK=LTAI5t... DEMO_ALIYUN_SK=xxxx STAGING_TENCENT_AK=AKID... STAGING_TENCENT_SK=xxxx

3. 核心功能实现：从取数到告警再到操作的完整链路

下面我把代码层面的实现拆开讲。这部分不追求把每个文件都贴出来，重点讲清楚几个容易出问题的环节是怎么处理的。

3.1 云资源采集：分页、状态归一化和多账号遍历

云厂商 OpenAPI 基本都需要分页拉取。拿阿里云 ECS 的 DescribeInstances 举例，每页最大返回 100 条，如果实例数超过 100，必须根据 TotalCount 循环翻页。

# collector/aliyun_ecs.py import json from aliyunsdkcore.client import AcsClient from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest def list_instances(access_key, secret_key, region_id): client = AcsClient(access_key, secret_key, region_id) request = DescribeInstancesRequest() request.set_PageSize(100) page = 1 instances = [] while True: request.set_PageNumber(page) response = json.loads(client.do_action_with_exception(request)) instances.extend(response.get("Instances", {}).get("Instance", [])) if page * 100 >= response.get("TotalCount", 0): break page += 1 return instances

这段代码本身不复杂，真正需要上心的是“状态归一化”。阿里云返回的是 Running、Stopped，腾讯云返回的是 RUNNING、STOPPED，AWS 返回的是 running、stopped——如果规则引擎直接拿原始字符串去匹配，同一套规则没法应用到多账号。CloddsBot 在采集层维护了一张映射表，把各厂商的状态统一为 running、stopped、starting、stopping、expired、error 几种枚举，规则引擎只认枚举值。

多账号遍历时，我会用并发限制来控制对云厂商 API 的请求速率。不是所有请求越快越好，很多账号默认 QPS 很低，几个区域同时拉很容易触发限流。实现上我用 ThreadPoolExecutor 加 max_workers=4，每个账号之间用独立令牌桶错峰，实测这个配置最稳。

3.2 规则引擎：阈值判断 + 时间窗口 + 告警抑制

规则引擎是 CloddsBot 的核心，我用 rules.yaml 描述规则，加载后变成内存中的规则对象。

# config/rules.yaml rules: - name: instance_down target: instance.status condition: "!= running" level: warning window: 15m - name: disk_usage_high target: instance.metrics.disk_usage condition: "> 85" level: critical window: 30m aggregate: true - name: cert_expire_soon target: cert.days_left condition: "< 30" level: warning window: 1d

这里有两个细节值得展开。第一是 window 字段，它是“事件去重的观察窗口”：同一台实例如果持续宕机超过 30 分钟，规则引擎不会每 5 分钟推一次告警，而是窗口内只保留第一条，避免告警风暴。第二是 aggregate 字段，它解决的是“多台实例同时异常”的合并问题。真实场景里，很多所谓宕机是机房网络抖动导致的一批实例同时不可达，逐台推送毫无意义，反而让人忽略真正重要的单点故障。CloddsBot 会在一次巡检中对同一区域内多次命中同一条规则的事件做聚合，合并成一条“该区域有 12 台实例状态异常，列表如下”的汇总消息。

聚合消息的效果大概是这样的：

【区域异常汇总】cn-hangzhou 共 12 台实例不可达 - i-bp1234abcd-web-01 - i-bp5678efgh-web-02 - i-bp9012ijkl-db-01 ... 首次检出时间：2024-06-01 10:23:00 当前状态：已持续 6 分钟

3.3 Webhook 推送：加签与消息模板

钉钉机器人支持三种安全设置，我建议直接用“加签”。签名逻辑是时间戳换行拼接密钥，用 HmacSHA256 计算后做 Base64 和 URL 编码，拼到 Webhook 地址上。

# notifier/dingtalk.py import time import hmac import hashlib import base64 import urllib.parse import requests def build_signed_url(webhook, secret): timestamp = str(round(time.time() * 1000)) string_to_sign = f"{timestamp}\n{secret}" hmac_code = hmac.new( secret.encode(), string_to_sign.encode(), digestmod=hashlib.sha256 ).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return f"{webhook}&timestamp={timestamp}&sign={sign}" def push_markdown(webhook, secret, title, text): url = build_signed_url(webhook, secret) payload = { "msgtype": "markdown", "markdown": {"title": title, "text": text}, } requests.post(url, json=payload, timeout=10)

消息模板我强烈建议用 Markdown 格式而不是纯文本。告警消息里至少包含：异常资源名称、账号 / 区域、当前状态、持续时间、可能的处理操作。因为群里消息是给人快速判断的，丢掉关键信息等于没推。推送失败时也要有兜底逻辑：requests 返回非 200 时，把告警事件标记为 pending，等下一轮巡检重试，而不是直接丢弃。

3.4 操作闭环：cloddsbotctl 与审计表

IM 推送只解决“发现问题”，真正“处理问题”我放在了 cloddsbotctl 这个命令行工具里。

cloddsbotctl status --account demo cloddsbotctl restart --account demo --instance-id i-bp1234abcd cloddsbotctl audit --last 50

为什么不用 Web 界面？因为一个 Web 控制面意味着要处理登录鉴权、会话管理、CSRF 这些问题，对内部工具来说维护成本不划算。SSH 到主机跑一条命令，反而是最透明、最容易审计的方式。每次执行操作都会往 SQLite 的 operations 表里写入操作人、目标资源、操作类型、结果状态。操作前还会先调用查询接口确认资源当前状态，避免对一台已经处于 stopping 状态的实例重复下发 restart，这种保护在云 API 的幂等性不够完善的场景下非常必要。

operations 表结构设计很简单：

CREATE TABLE operations ( id INTEGER PRIMARY KEY AUTOINCREMENT, operator TEXT NOT NULL, account TEXT NOT NULL, resource_id TEXT NOT NULL, action TEXT NOT NULL, status TEXT NOT NULL, message TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) );

3.5 数据存储：为什么 SQLite 就够用

CloddsBot 的事件、告警、操作记录数据量一天最多几百条，一年也不到十万条，SQLite 单文件完全能扛住。用 SQLite 还有一个好处是备份简单——定时把 db 文件复制到对象存储即可，不需要搭数据库服务。唯一需要留意的点：多线程写入时 SQLite 可能遇到 database is locked，解决方案是开启 WAL 模式并让写入走单一线程。

import sqlite3 conn = sqlite3.connect("data/cloddsbot.db") conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA synchronous=NORMAL;")

开启 WAL 之后，读和写可以并行，日常使用基本不会碰到锁问题。

4. 部署落地：容器化、Systemd 守护与参数调优

4.1 Docker 镜像与 compose 配置

先看 Dockerfile，我基于 python:3.11-slim，安装依赖后直接运行主程序：

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "main.py"]

docker-compose.yml 中需要特别注意两点：挂载配置目录和数据库目录到宿主机，设置 restart: always。

services: cloddsbot: build: . restart: always volumes: - ./config:/app/config:ro - ./data:/app/data env_file: - .env

配置目录只读挂载，防止进程运行时意外修改配置文件；数据目录持久化，这样容器重建后历史事件和审计记录不会丢。还有个细节：容器时区默认是 UTC，如果不想在代码里处处处理时区，可以在 compose 里加 environment 的 TZ=Asia/Shanghai，但记得代码里的调度表达式还是显式指定 timezone 最保险。

4.2 Systemd 管理而不是裸跑 Docker

常驻服务最好交给 Systemd 托管。下面是 unit 文件要点：

[Unit] Description=CloddsBot runtime After=docker.service Requires=docker.service [Service] WorkingDirectory=/opt/cloddsbot ExecStart=docker compose up ExecStop=docker compose down Restart=always RestartSec=10

这样开机自启、崩溃重启、日志统一 journalctl 管理。日志轮转方面可以加一条 logrotate 配置，对 /var/log/cloddsbot 目录按天切割并保留 30 天。主程序内部我用 logging 模块同时输出到 stdout 和文件，stdout 走 journald，文件走 logrotate，两边都不耽误。

4.3 实测运行时参数与调度频率

运行三个月后的经验是：主程序常驻内存约 120MB，每次巡检耗时在几秒到十几秒之间，主要取决于账号数量和实例规模。调度频率建议是一个折中方案——实例状态检查每 5 分钟一次，证书到期检查每天一次，账单异常检查每天一次（账单数据本身 T+1 才更新，查太频繁没有意义）。个人经验是把“实时性要求高”的状态检查和“低频但重要”的到期检查分开配置，而不是一刀切全按 5 分钟跑。

5. 运行三个月后回头看：那些坑和对应的解法

5.1 最开始的告警风暴：一次误报带来几十条推送

第一个版本没有告警抑制，某天凌晨区域网络抖动，我同时收到几十条“实例状态异常”的钉钉消息。那次之后我做了两个改动：一是引入上文提到的 window 去重，二是对同一批次异常做聚合。还有一个心态层面的收获：告警推送的“权威性”很宝贵，如果群里经常刷无意义告警，团队成员会逐渐无视所有通知，那才是最大的失败。

排查链路值得记录一下：当时我先去看采集日志，发现所有不可达实例都集中在同一区域，而且时间点完全一致；然后单独调用一次该区域查询接口，发现返回正常；再翻外层网络监控，发现是机房出口抖动。整个过程其实不到十分钟，但因为消息刷屏，群里所有人都被惊动了。从那以后，凡是不涉及具体资源 ID 的异常，我都默认先聚合再推送。

5.2 云厂商 SDK 的分页参数并不一致

这一点在跨厂商接入的时候非常坑。阿里云用 PageNumber/PageSize，腾讯云部分接口也用类似风格，但 AWS 风格接口用的是 NextToken 游标。写统一 Provider 接口时，建议把“分页遍历”封装成生成器，上层调用方只负责一个个拿资源，不感知底层差异。

def iter_instances(provider, account, region): for raw in provider.list_instances(account, region): yield normalize_instance(raw)

当时我踩的坑是：某个厂商的 SDK 默认每页只返回 20 条，而它的 SDK 文档里关于分页的说明藏得很深。第一周巡检出 25 台实例只推了 20 台，剩下 5 台的状态完全没有被检查到。这个问题后来是通过一个“资源总数自检”发现的——拉取列表后把归一化后的实例数和厂商返回的 TotalCount 做比对，不一致就直接告警“采集可能不完整”。这个自检逻辑强烈建议保留。

5.3 时区问题永远是定时任务的隐形杀手

我最初把所有调度表达式都写成本地时间，后来添加了海外区域资源后，发现告警里的时间戳对不上。最终统一约定：程序内部所有时间存储使用 UTC，仅在与用户交互展示时转换成本地时间；APScheduler 的 cron 表达式全部显式指定 timezone=Asia/Shanghai。不要在代码里隐式依赖系统时区，否则部署到不同环境很容易出现“凌晨三点没跑巡检”这种诡异问题。

还有个小坑：SQLite 的 datetime('now') 返回的是 UTC 时间，如果插入时不注意，查出来的时间会和你本地时间差 8 小时。所以我在表里存时间统一用 ISO 8601 带时区后缀，或者干脆存 UTC 并在查询时显式转换。

5.4 SSL 证书检测的翻车经历

实现证书到期提醒时，我最初只检查了证书的 notAfter 字段，结果一条生产环境的证书在到期前一周就被我判成“正常”。原因是该证书链里的中间证书先过期了，实际握手已经失败。所以证书检查不能只看叶子证书的到期日期，有条件的话应该直接用 openssl 做一次真实握手来验证。

echo | openssl s_client -servername api.example.com -connect api.example.com:443 2>/dev/null | openssl x509 -noout -dates

这也是“状态巡检类工具”最容易犯的错误：以为字段齐了就是对的，其实数据真实含义还得往下一层看。证书到期时间字段存在不代表证书链完整，磁盘使用率字段存在不代表云盘没有进入只读状态。规则引擎只能帮你发现你定义过的问题，定义问题本身需要你对底层机制有足够的了解。

5.5 机器人 Webhook 地址泄露后的补救

自定义机器人 Webhook 如果泄露，任何人都能往你的群里推消息。钉钉后台可以重置密钥，但更稳妥的做法是：把 Webhook 和密钥放到环境变量里统一管理，在配置中不要写明文；同时给推送接口设置一个“每日消息数上限”，异常流量直接被熔断，防止群被刷屏。

MAX_DAILY_MESSAGES = 200 def can_send_today(): count = db.execute( "SELECT count(*) FROM notifications WHERE date(created_at) = date('now')" ).fetchone()[0] return count < MAX_DAILY_MESSAGES

其实这个限制还有一个额外收益：它会逼着你优化告警质量。消息配额有限，你自然会去合并同类项、去除低级别噪音，最后群里剩下的都是真正值得看的。

6. 这个项目接下来我打算怎么演进

CloddsBot 目前的状态已经能覆盖我绝大部分日常巡检需求，但后续还有几个明确想做的方向。现阶段每加一个新能力都遵循同样的套路：先想清楚数据从哪来，再想清楚异常怎么定义，最后才写代码。

6.1 费用异常检测

云厂商账单接口支持拉取按产品线分组的日账单。思路是在账单异常规则里加“对比近 30 天均值”的判定，比如某产品线当天费用超过前 30 天日均费用的 3 倍时告警。这类场景用规则引擎的框架同样能覆盖，只是数据源从实例列表换成了账单接口。

费用异常检测有一个特殊点：账单数据是 T+1 的，所以不需要高频巡检，但需要保留足够长的历史数据做基线。Sqlite 直接存 30 天账单原始值，用 window 函数取均值即可，不需要额外引入时序数据库。

6.2 多账号操作审批流

现阶段 cloddsbotctl 只有本机审计，没有审批。如果后面加入“推送消息 → 指定负责人确认 → 确认后执行”的流程，就需要一个能暴露给 IM 回调的轻量服务。我的初步想法是做一个只监听内网 HTTP 的确认服务，配合企业自建应用的回调能力，可以大幅降低误操作风险。

审批流的设计原则是“最小暴露面”：确认服务不挂在公网，只在内网监听；每个确认请求带一次性 token，30 秒过期；操作执行前后各写一条审计。这种内部工具不追求功能花哨，稳定和可追溯永远排在第一位。

6.3 巡检报告周报

把每周的事件和告警做汇总，生成一份 Markdown 周报定时推到群里。内容包括异常类型分布、处理耗时、当周新增资源数。这种周报对个人开发者可能无所谓，对小团队来说是很好的透明化手段。

周报模板大致长这样：

## 本周巡检报告（06.01 - 06.07） 本周共巡检 8 个账号、137 台实例，发现 23 个异常事件。 | 异常级别 | 数量 | 已处理 | | --- | --- | --- | | critical | 2 | 2 | | warning | 21 | 19 | 异常类型分布： - instance_down：1 - disk_usage_high：3 - cert_expire_soon：17 - cost_anomaly：2

如果你也正被“每天登录控制台巡检”这种低效习惯困扰，不妨照着上面的思路拆一个自己的版本。核心逻辑其实就三步：拉数据、定规则、推消息。真正花时间的从来不是代码，而是想清楚“什么才算异常”以及“推给谁看最有价值”。