拓冰网站定制 
做爬虫的朋友大概率都有过这种经历凌晨两三点某个关键采集任务挂了一整夜第二天早上打开后台——企业微信和钉钉的告警群安安静静而数据库里该入库的数据一条没进来。最难受的还不是修代码而是你不知道它在几点挂掉的、中间发生了什么。给爬虫配上监控告警让异常在发生的第一时间变成手机上的实时预警这件事我是被几次深夜事故逼着认真做起来的。这篇文章就围绕怎么结合企业微信或钉钉打造一套爬虫专用的7×24小时实时预警系统来讲。我会把告警渠道选型、机器人接入细节、监控指标怎么定、告警触发策略怎么设计以及一套可以直接改改就用的Python告警模块全部梳理清楚。不管你在跑的是requests脚本、Scrapy项目还是分布式采集任务把这里的思路复制到自己的环境里基本都能用得上。1. 深夜静默失败我决定给爬虫装告警的那个夜晚1.1 爬虫悄无声息挂掉的几种典型死法爬虫不会像Web服务那样挂了就返回500。它最大的特点是程序还活着但采集质量已经严重下降了。我总结了几种最常见的静默死亡场景你可以对照一下自己有没有遇到过。第一种是目标站点页面改版。这是最普遍的死法。选择器失效、字段位置变动、翻页参数变化程序只负责发请求和解析页面结构一变解析结果直接变0。重点是这个过程中请求成功率依然很高日志也不会报错看起来一切正常实际上已经白跑了一整夜。第二种是代理IP池失效。做采集的基本都离不开代理尤其是对访问频率敏感的目标源。代理池里的IP一个个失效程序不断重试最终大量请求积压超时。有些任务配置了重试机制碰到失败就一直重试结果一个错误请求被放大成几十次日志刷屏也不报警。第三种是触发了反爬机制但程序没识别出来。服务器返回200Response内容却是验证码、跳转页或风控提示。如果解析逻辑里没有对这类内容做判断程序就会把垃圾页面当成正常数据来源解析字段全部为空。这种故障隐蔽性极高也是最难排查的。第四种是基础设施问题。磁盘写满、内存不足、服务被系统OOM杀掉或者分布式爬虫的某个Worker节点悄悄断连。任务队列里的消息没有消费者整个流程堵在那里但主进程还在运行看起来一切正常。第五种是外部依赖异常。比如目标数据源需要登录态而登录态的Cookie过期了或者爬虫依赖的某个接口调整了鉴权方式导致调用直接403。这类问题往往发生在凌晨因为很多站点会在低峰期更新服务器配置。1.2 没有告警时损失是被一步步放大的没有监控告警的时候一次静默故障的实际损失远不止几小时的数据缺失。第一步是数据断档。很多采集任务对时效性要求很高比如商品价格、股票行情、舆情信息错过的时间窗口补不回来。就算允许补采几小时的数据积压也够你追很久。我曾经有一个任务挂了6个小时恢复之后跑了整整一天才把积压数据补完期间新数据又要继续采两头挤压。第二步是信任损耗。爬虫跑出来的数据往往是给业务方、给下游系统用的。数据延迟一次人家能理解延迟多了整个采集系统的口碑就没了。每次都要你主动解释昨晚又挂了这种体验真的很消耗个人信誉。第三步是封禁风险被放大。很多情况下故障往往伴随着对目标站点的异常请求模式。比如代理失效后重试风暴或者登录态失效后反复请求异常接口这些行为都会加重风控识别。本来及时发现、及时停手就能避免的封禁因为没有告警反而让反爬系统看到了一个死而不僵的客户端处理起来更麻烦。所以告警系统不是为了在出事的时候通知一声那么表面它的本质是缩短故障发现时间。能把几个小时的故障发现时间压缩到几分钟损失的放大效应就被控制住了。这也是为什么我一直坚持爬虫的监控告警不是可选项它是整套采集系统里和爬虫代码同等重要的一部分。2. 告警渠道选型为什么最后留下了企业微信和钉钉2.1 邮件、短信和IM在告警场景下的实际体验差异告警消息要送到人手上渠道很关键。我最初用的是邮件后来又试过短信最后才把重心放到企业微信和钉钉这类IM机器人上。这里面的取舍很有意思。渠道到达时效格式能力成本被打断感适合场景邮件分钟级且容易进垃圾箱富文本强但手机上查看并不及时低很弱日报汇总、非紧急通知短信秒级但常有延迟和丢失纯文本无法排版按条收费很强电话无人接听时的兜底IM机器人秒级支持Markdown、引用、人免费可控实时告警、协作处理邮件的核心问题不是送不到而是送达不等于看见。人不会一直盯着邮箱尤其是凌晨。短信虽然能推但纯文本的格式太受限一条长告警发过来根本看不完。而企业微信和钉钉机器人几乎是专为这类场景设计的免费、秒达、能发结构化消息还能在群里责任人。另外说一句现在很多监控系统比如夜莺这类开源工具的告警通知也默认支持企业微信和钉钉机器人这说明IM渠道已经是社区验证过的主流方案了。你要做的不是从零发明而是把这类通知能力接入到自己的爬虫监控体系里。2.2 企业微信机器人适合已经用微信协作的团队企业微信机器人最大的优势是离人近。国内很多团队的工作沟通就在企业微信上告警消息推到群里大家顺手就看到了不需要额外打开App。从技术角度看企业微信群机器人的接入也足够省事在一个群里添加自定义机器人拿到Webhook地址然后往这个地址POST一个JSON消息就行。消息格式支持文本、Markdown、图片、图文链接等。Markdown对告警来说完全够用标题、加粗、链接、引用都能渲染。它的另一个特点是支持在消息里人。你可以通过mentioned_list传用户的userid也可以通过mentioned_mobile_list传手机号来指定的人。这个能力在告警分级里非常实用P1故障直接值班工程师而不是所有人手机都在响。限制方面企业微信群机器人官方对单个机器人有频率限制一般认为是每分钟20条左右。单看这个数字感觉不小但如果告警策略没做好凌晨一小时触发几百条直接触发限流事小把告警群变成垃圾群导致大家屏蔽通知才是大问题。这个我在后面的告警策略部分会重点讲。2.3 钉钉机器人安全设置更硬核钉钉机器人放在同一维度比较接入流程和企业微信很像建群、加自定义机器人、拿Webhook、POST消息。但它有几个差异点值得注意。首先是安全设置更丰富。钉钉自定义机器人必须选择一种安全设置自定义关键词、加签、IP白名单三种可以多选。自定义关键词要求推送的消息里必须包含指定关键词否则拒绝发送加签要求对请求做HMAC-SHA256签名IP白名单则只允许指定来源IP调用。相比企业微信钉钉的这套机制对防止Webhook地址泄露后被人乱刷的保护更完善。其次是消息类型。钉钉机器人支持text、markdown、link、actionCard、feedCard等类型。其中markdown类型同样支持标题、加粗、引用、列表告警场景完全够用。而actionCard类型可以推送一张带有按钮的卡片点击按钮跳转到指定链接这个功能用来做点击处理告警很顺手。选企业微信还是钉钉其实不用太纠结。看你们团队日常用哪个办公IM就选哪个如果两边都在用也可以同时接入做成双通道冗余。比如企业微信为主钉钉为备份通道这样万一一个IM服务出问题告警还能从另一个渠道送达。3. 企业微信机器人接入建群、Webhook、加签校验一条龙3.1 建群添加机器人的关键步骤与权限坑实际接入企业微信机器人差不多五分钟就能搞定。操作路径是在企业微信里创建一个群聊哪怕只有你自己一个人也行然后进入群设置找到群机器人选择添加机器人再选新创建一个机器人。给它起个名字比如爬虫告警创建成功后就能看到Webhook地址。这里面有几个坑我一个个说。第一只有群主和群管理员才有权限添加机器人。如果你在一个别人的群里没有管理权限是找不到添加机器人入口的。建议每个项目单独建一个告警群并由实际负责维护的人做群主。第二机器人创建后Webhook地址里包含一个key参数这个key就相当于机器人的密钥。不要把它写进会被提交到Git仓库的配置里更不要截图发到公开渠道。我的习惯是放到环境变量或独立的配置文件中并且配置文件的读取权限设成只有启动用户可读。第三如果机器人被移出群聊Webhook地址会失效所有调用都会报错。碰到这种情况重新创建一个机器人即可代码里更新一下地址。创建机器人的时候还可以选择加签安全设置。企业微信新版机器人支持配置一个签名密钥发送请求时需要在URL里带上计算的签名。这块我下一步详细说。3.2 Webhook消息格式与加签校验代码企业微信机器人发送消息时HTTP请求的Body是一个JSON。以Markdown消息为例最基本的格式是这样{ msgtype: markdown, markdown: { content: ## 爬虫预警\n 任务price_spider\n错误信息连续3个周期解析结果为空 } }如果配置了加签需要在Webhook地址后面拼接timestamp和sign两个参数。加签的算法逻辑如下取当前时间的Unix秒级时间戳拼字符串timestamp \n secret以secret为密钥对上述字符串做HMAC-SHA256摘要将摘要结果做Base64编码得到sign。下面是一段完整的Python实现包含加签URL生成和消息发送import base64 import hashlib import hmac import time import urllib.parse import requests class WecomNotifier: def __init__(self, webhook_url: str, secret: str ): self.webhook_url webhook_url self.secret secret def _signed_url(self) - str: if not self.secret: return self.webhook_url timestamp str(int(time.time())) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign base64.b64encode(hmac_code).decode(utf-8) # 实测发现 base64 结果里的 / 直接拼进 URL 可能被截断建议做一次编码 sign urllib.parse.quote_plus(sign) connector if ? in self.webhook_url else ? return f{self.webhook_url}{connector}timestamp{timestamp}sign{sign} def send_markdown(self, content: str) - bool: payload { msgtype: markdown, markdown: {content: content}, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: print(fsend failed: {result}) return False return True调用方式很简单wecom WecomNotifier( webhook_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyYOUR_KEY, secretYOUR_SECRET, ) wecom.send_markdown(## 爬虫预警\n 采集任务异常请检查。)实测经验签名校验失败时企业微信会返回类似invalid sign的提示。最容易踩的坑是服务器时间不同步导致本地生成的timestamp和服务端时间差太多签名直接不通过。上线前先确认运行告警模块的服务器时间是用NTP同步过的。3.3 企业微信推送的限流和格式限制企业微信机器人的限制主要是频率和Markdown语法范围。频率上单个机器人每分钟大概能推送20条左右超过后服务端会返回限流错误。这个限制对正常告警来说够用但前提是不能在告警策略里做每5秒推一条这种设计。Markdown格式方面企业微信支持的是其特定子集#标题、**加粗、引用、[文本](链接)、无序列表、有序列表、font color颜色设置等。不支持的语法会自动降级显示。实测中|表格语法在企业微信里是不会渲染成表格的这点和钉钉有差异。所以我在写告警模板时习惯用标题加引用的方式排版而不是用表格。推送内容上我建议一条消息不要超过2000字。手机端消息太长会被折叠反而抓不住重点。告警内容要摘要化什么任务、什么级别、什么现象、谁负责、处理入口这几个要素就够了细节留给日志系统去查。4. 钉钉机器人接入加签、关键词、IP白名单的取舍4.1 三种安全设置怎么选钉钉自定义机器人的安全设置可以单选或多选。三种方式的适用场景差别很大我列个表对比一下安全方式原理优点缺点推荐场景自定义关键词消息文本必须包含指定关键词配置简单关键词被猜中后可被伪造需要消息文本包含关键词内网告警追求低成本接入加签请求URL带HMAC签名安全性最高无法简单伪造需要在代码里实现签名逻辑生产环境默认推荐IP白名单只允许指定来源IP调用Webhook直接屏蔽外部IP如果服务器IP变化或走代理需要同步维护固定服务器部署我的建议很明确只要不是临时测试一律用加签。加签相当于给Webhook地址加了一把只有你的代码能打开的锁即使地址泄露别人也推不了消息。IP白名单可以当作辅助手段叠加使用但如果你的爬虫部署在多个动态IP节点上维护成本会比较高。4.2 钉钉Markdown消息格式与发送代码钉钉的Markdown消息格式和企业微信有些区别。钉钉机器人对Markdown的支持标题用####这种四级标题效果比较好看引用、**加粗、-列表都是支持的但不支持表格链接语法支持。下面是一段完整的钉钉加签发送实现import base64 import hashlib import hmac import time import urllib.parse import requests class DingTalkNotifier: def __init__(self, access_token: str, secret: str ): self.access_token access_token self.secret secret self.base_url https://oapi.dingtalk.com/robot/send def _signed_url(self) - str: timestamp str(round(time.time() * 1000)) # 注意钉钉用毫秒时间戳 url f{self.base_url}?access_token{self.access_token} if self.secret: string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) url ftimestamp{timestamp}sign{sign} return url def send_markdown(self, title: str, text: str, at_mobiles: list None) - bool: payload { msgtype: markdown, markdown: { title: title, text: text, }, } if at_mobiles: payload[at] { atMobiles: at_mobiles, isAtAll: False, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: print(fsend failed: {result}) return False return True用法ding DingTalkNotifier( access_tokenYOUR_ACCESS_TOKEN, secretSECYOUR_SECRET, ) ding.send_markdown( title爬虫预警, text#### 爬虫预警\n\n 任务price_spider\n\n**状态**连续3个周期解析结果为空\n\n[查看详情](https://your-monitor.example.com), at_mobiles[13800138000], )这里特别提醒一下钉钉加签的时间戳是毫秒级不是秒级。我之前用秒级时间戳签名验证就没通过排查了好久才发现是单位搞错了。另外加签后生成的sign必须用quote_plus编码因为Base64编码结果里可能包含、/、这些URL特殊字符不编码会被服务端解析错误。4.3 钉钉限流、错误码和重试策略钉钉自定义机器人的限流策略也比较严格。单个机器人每分钟最多推送大约20条如果调用过频繁服务端会返回限流错误。另外钉钉还会校验消息内容中是否包含匹配的安全设置比如你选择了关键词方式但消息里没有关键词会返回类似keywords not in content的错误错误码一般是310000一类的提示。处理这类失败时我建议在发送函数里做重试但要加退避。愚蠢的重试方式是失败后立刻再发一次这在凌晨告警场景里很可能把服务端限流打到更严重。我用的策略是第一次失败后等待2秒重试再失败就等5秒最多重试3次仍失败则放弃本次推送并把告警写入本地文件。宁可少推送一次也不要因为重试风暴把群里的消息渠道彻底搞挂。另外钉钉机器人发送失败并不都是网络问题也可能是消息格式不合法。这类问题重试多少遍都没用需要检查返回的errmsg提前在本地拦截。比如Markdown格式里的语法错误、文本长度超过限制这些在发送前就可以规避。5. 监控指标设计先定义值得半夜叫醒你的异常5.1 从日志中能提炼出的核心指标告警系统不能什么都报否则跟不报没区别。要设计好告警首先是定义监控指标。我把爬虫运维里最常用的核心指标整理成了下面这张表指标含义异常场景示例请求成功率成功请求数 / 总请求数代理失效、目标接口临时故障有效解析率成功解析出目标字段的页面占比页面改版、数据结构变更提取数据量单位时间窗口内入库的记录数数据积压、采集逻辑阻塞平均响应时间 / P95请求耗时的集中趋势和长尾情况带宽问题、目标源限速代理池可用率可用代理数 / 总代理数代理供应商故障、账号欠费队列积压量待采集任务队列里的消息数量Worker节点断连、消费能力下降进程存活状态爬虫主进程是否持续运行内存溢出、崩溃退出磁盘与内存使用率主机基础资源水位日志吞盘、数据未及时归档这些指标听起来多实际上都很好采集。最简单的做法是让爬虫把每次请求的关键信息写进结构化日志JSON格式的一行然后由监控程序定时读取日志窗口做聚合。稍微复杂一点的可以把指标上报到Prometheus再用Grafana展示。但告警模块本身不一定要依赖整套监控系统——对大多数脚本型爬虫来说一个能跑定时任务的Python进程就够了。5.2 单指标容易误判联动才是关键设计监控指标最容易犯的错误是只看单一指标。我来举个例子请求成功率下降了但有效解析率和数据量都正常。这大概率只是个别接口波动不影响大局不需要大呼小叫。反过来请求成功率很高但有效解析率是0这才是真正要命的故障——说明页面结构变了或反爬页面混进了正常请求里。所以我做告警判断时基本都会采用组合条件。比如连续3个统计周期每个周期5分钟内有效解析率低于50%并且提取数据量低于正常基线的30%才触发解析异常告警。任务队列积压量超过阈值同时活跃Worker数低于预期才触发分布式节点故障告警。请求成功率低于90%并且平均响应时间超过基线的两倍才触发网络链路异常告警。组合条件的逻辑很简单单个指标的波动可能有很多原因但多个指标同时反常往往意味着真实故障。这样做还有一个额外的好处是可以大幅减少误报——误报多了人就会疲于应对最后连真告警也没人看。5.3 阈值不要拍脑袋先跑一周基线阈值怎么定是最考验经验的地方。我的建议很简单不要在系统刚上线的时候拍脑袋定阈值先让它“只记录、不告警”地跑上一周把正常运行的指标分布收集起来再回头定阈值。比如页面解析成功率正常情况可能是99.5%左右偶尔跌到90%但很快恢复。如果你直接把阈值定在99%那这一周的指标会让你发现原来某些低峰时段本来就会有一些波动。合理的阈值应该设在“正常波动范围之外但又不至于等故障扩大到不可挽回才触发”的位置。具体方法上我个人喜欢用百分位数。比如提取数据量取过去一周每个5分钟窗口的P5和P95那么低于P5-20%或高于P9520%都可以视为异常候选。这种方法比单纯设一个绝对数字更贴近实际。当然每个业务不一样绝对阈值也有用比如磁盘使用率超过85%这类资源指标就和业务量无关直接定绝对值就行。6. 告警触发策略静默期、恢复通知、分级升级的设计6.1 阈值告警和趋势告警要怎么配合告警触发策略如果是超过一个固定阈值就推消息实现简单但容易漏报慢变故障。比如代理池可用率从80%慢慢降到30%过程中可能好几个小时都没有一次触发阈值等你发现时代理池已经崩了。所以在实践里我会把阈值告警和趋势告警结合起来用。趋势告警的方式不复杂每个统计周期都计算当前指标与历史基线的偏离程度一旦偏离持续扩大即使还没到绝对阈值也先发一条低级别告警提示。举个例子请求成功率当前98%还没到95%的硬阈值但和过去三天的同期水平相比下降了5个百分点并且连续两个周期都在下滑这时就应该推一条P2告警让人提前介入。这样做的好处是很多故障被消灭在刚冒头的阶段。代价是趋势告警需要有一个稳定的历史基线数据所以基线收集这一步不能跳过。6.2 静默期和去重防止凌晨的告警风暴告警系统的最大敌人是告警风暴。凌晨2点一个故障触发如果策略写得粗可能每隔两分钟就推一条相同告警。结果就是手机一晚上响个不停第二天群里全是重复消息所有人对告警产生免疫真正的严重故障反而被淹没。解决思路就是静默期和去重。静默期指的是同一类告警在指定时间内只允许推送一次比如600秒内同一个task_id的解析异常不再重复推送去重则是把多条同类异常聚合成一条消息附带最近10分钟内累计失败N次这样的汇总信息。我实现的告警管理器里会维护一个字典记录每个告警key的最后推送时间只有超过静默期才允许再次推送。具体的实现代码我会在下一章给出。这里先强调一个原则告警的目的是让人知道出了事和事有多严重不是让系统实时播报故障的每一个细节。细节属于日志系统和监控面板。6.3 分级与升级规则让通知更有轻重缓急告警必须分级不然后果就是P1和P3用同一种方式轰炸所有人。我常年的分级维度如下P1严重采集任务已停止、关键数据源不可用、代理池整体失效、磁盘满导致无法写入。这一类告警应立即推送并值班人。P2主要指标明显偏离但业务未完全中断比如解析率下降、响应时间飙升、队列积压增长。这类告警推送即可不需要所有人但需要人工确认。P3次要单次超时、夜间低峰期的一些小波动、个别页面解析失败。这类场景不建议实时推送而是汇总成日报白天统一看。升级规则也很重要。我的策略是P3持续超过1小时自动升级为P2P2持续超过30分钟仍未恢复自动升级为P1并追加人。升级的本质是给故障一个被处理的机会人不回不看告警就会自己往更高层级走直到有人接手。7. 实战代码一个同时兼容企业微信和钉钉的告警模块7.1 通用告警消息结构定义既然企业微信和钉钉都要接入那就先把告警消息抽象成一个数据结构收发双方都依赖这个结构。这样可以保证切换渠道时告警内容的组装逻辑不用改动。我用dataclass定义一个AlarmMsg它包含标题、内容、级别、任务ID、对象和时间戳等字段from dataclasses import dataclass from typing import List, Optional dataclass class AlarmMsg: title: str content: str level: str P2 # P1 / P2 / P3 task_id: str at_mobiles: Optional[List[str]] None # 钉钉 手机号 at_userids: Optional[List[str]] None # 企业微信 userid timestamp: str 然后在组装消息时统一由这个对象出发企业微信和钉钉的Notifier各自负责把它渲染成对应平台的Markdown格式。这样做的好处是以后就算要加一个飞书机器人也只需要再写一个Notifier类业务代码几乎不用动。7.2 企业微信Notifier实现结合前面讲到的加签逻辑我把企业微信的Notifier封装成一个类import base64 import hashlib import hmac import time import urllib.parse import requests class WecomNotifier: def __init__(self, webhook_url: str, secret: str ): self.webhook_url webhook_url self.secret secret def _signed_url(self) - str: if not self.secret: return self.webhook_url timestamp str(int(time.time())) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) connector if ? in self.webhook_url else ? return f{self.webhook_url}{connector}timestamp{timestamp}sign{sign} def send(self, msg: AlarmMsg) - bool: text f## {msg.title}\n text f 级别**{msg.level}**\n if msg.task_id: text f 任务{msg.task_id}\n text f时间{msg.timestamp}\n\n text msg.content payload { msgtype: markdown, markdown: {content: text}, } if msg.at_userids: payload[mentioned_list] msg.at_userids resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() return result.get(errcode) 0这里有个细节企业微信的mentioned_list需要传成员UserID如果你的账号体系里拿不到UserID可以用mentioned_mobile_list传手机号两者效果类似。7.3 钉钉Notifier实现钉钉的Notifier实现逻辑很接近区别在于时间戳用毫秒、签名要拼在access_token后面class DingTalkNotifier: def __init__(self, access_token: str, secret: str ): self.access_token access_token self.secret secret self.base_url https://oapi.dingtalk.com/robot/send def _signed_url(self) - str: timestamp str(round(time.time() * 1000)) url f{self.base_url}?access_token{self.access_token} if self.secret: string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) url ftimestamp{timestamp}sign{sign} return url def send(self, msg: AlarmMsg) - bool: text f#### {msg.title}\n\n text f 级别**{msg.level}**\n if msg.task_id: text f 任务{msg.task_id}\n text f时间{msg.timestamp}\n\n text msg.content payload { msgtype: markdown, markdown: { title: msg.title, text: text, }, } if msg.at_mobiles: payload[at] { atMobiles: msg.at_mobiles, isAtAll: False, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() return result.get(errcode) 0钉钉的消息text里不支持企业微信里那种##一级标题的渲染效果我实测####展示出来更干净。这个差异属于平台特性写模板时要分开处理。7.4 集成到爬虫主循环监控-判定-发送的完整链路有了Notifier还差一个控制静默期的管理器。我用一个简单的类来实现去重逻辑import time class AlarmManager: def __init__(self, notifier, cooldown_seconds: int 600): self.notifier notifier self.cooldown_seconds cooldown_seconds self._last_sent {} def should_alarm(self, key: str) - bool: now time.time() last self._last_sent.get(key, 0) if now - last self.cooldown_seconds: return False self._last_sent[key] now return True def send(self, msg: AlarmMsg, key: str ) - bool: key key or msg.title if not self.should_alarm(key): return False return self.notifier.send(msg)然后把它接到爬虫主循环里。下面是一个简化但完整的示例演示了抓取-计算指标-判断异常-告警的全链路def crawl_once(): # 这里替换成你的真实采集逻辑 stats { request_success_rate: 0.98, parse_success_rate: 0.30, item_count: 120, task_id: price_spider, } return stats def run_monitor(): notifier WecomNotifier(WECOM_WEBHOOK_URL, WECOM_SECRET) alarm AlarmManager(notifier, cooldown_seconds600) while True: try: stats crawl_once() # 组合条件判定解析率低 数据量少 - P1 if stats[parse_success_rate] 0.5 and stats[item_count] 300: alarm.send( AlarmMsg( title采集任务疑似解析失效, content- 有效解析率{:.2%}\n- 提取数据量{}\n- 建议检查目标站点页面结构。.format( stats[parse_success_rate], stats[item_count] ), levelP1, task_idstats[task_id], timestamptime.strftime(%Y-%m-%d %H:%M:%S), ), keyfparse-fail:{stats[task_id]}, ) except Exception as exc: alarm.send( AlarmMsg( title采集任务异常退出, contentf异常信息{exc}, levelP1, task_idprice_spider, timestamptime.strftime(%Y-%m-%d %H:%M:%S), ), keyspider-exception, ) time.sleep(30) if __name__ __main__: run_monitor()这个环节里最核心的部分就是AlarmManager的静默期判断。没有它一个故障在600秒内反复触发20次你的手机就会被消息淹没。有了它同类告警每10分钟最多推一条既不会漏也不会吵。8. 部署与运维让监控系统自己也要24小时在线8.1 用systemd守护爬虫和监控进程监控模块本身也是一个进程它也有挂掉的风险。所以我会用systemd来守护爬虫任务和告警监控进程而不是简单地在终端里nohup一下。下面是我常用的一个systemd unit配置示例[Unit] DescriptionSpider Alarm Monitor Afternetwork-online.target [Service] Userspider WorkingDirectory/data/spider ExecStart/usr/bin/python3 /data/spider/alarm_monitor.py Restartalways RestartSec10 EnvironmentFile/data/spider/.env [Install] WantedBymulti-user.target配置好以后依次执行systemctl daemon-reload、systemctl enable spider-alarm、systemctl start spider-alarm就能开机自启。Restartalways是关键进程异常退出后10秒内会自动拉起。环境变量通过EnvironmentFile引入避免把密钥直接写在代码里。8.2 给监控系统加心跳长时间没消息就是最大的告警监控系统最尴尬的一个坑是监控进程挂了之后大家都不知道。因为没人收到告警这件事本身不会产生告警。我解决这个问题的方法是给监控系统加心跳机制。最简单的方式是每天固定时间在告警群里推一条汇总消息内容包括今日任务总数、成功任务数、失败任务数、告警触发次数。这条消息本身就是一个健康信号——如果某天早上没收到日报说明监控进程可能挂了。更进一步的做法是在多个核心采集节点上部署互看的探针。比如A节点的监控模块定时向B节点的Webhook地址发一个存活探测请求B节点收到后记录如果连续几次没收到A的心跳B就替A发出一条告警。这套互看机制不复杂但对监控系统自身的可用性能带来很大的提升。8.3 几个隐蔽但致命的运维坑最后分享几个我实际运维中踩过或帮别人排查过的坑都很隐蔽但遇到了都挺折腾。第一个是服务器时间不同步。前面已经提过一次这里再强调无论企业微信还是钉钉加签都依赖请求时间与服务器时间接近。如果服务器时间偏差超过一两分钟签名校验就会失败。我一般会在服务器上配置NTP自动同步并在告警模块启动时先校验一次本机时间。第二个是Webhook地址泄露。有一次我发现告警群里莫名其妙出现一些不是我写的测试消息排查后发现是Webhook地址被同事贴到公开文档里了。处理办法有两种在企业微信端删掉机器人重新创建一个在钉钉端开启IP白名单把无关来源直接挡在门外。平时养成习惯Webhook地址和secret只放进部署服务器的环境变量里。第三个是告警模块与爬虫部署在同一台机器磁盘满时一起挂。爬虫最占磁盘的往往是日志和抓取缓存一旦把盘写满爬虫和监控进程都会因为写日志失败而异常。我会把监控模块的日志单独写到另一个分区同时给日志目录配置按大小轮转避免单个日志文件无限增长。第四个是手机端的通知权限。企业微信和钉钉的群消息默认是免打扰的如果告警群没有开启关注的消息提醒消息倒是送到了但手机状态栏不弹提醒等于白搭。所以建完告警群第一件事就是把群设置里的消息通知改成关注的人或指定消息提醒并把机器人的消息类型设为关注项然后拿测试消息实际验证一遍手机能不能响。这些坑单看都不大但叠加在一起就足以让一套架构上看起来很完美的告警系统在关键时刻变成了摆设。我的经验是告警系统上线之后不急着加更多指标先花一周时间把消息能否稳定到达手机这件事反复验证清楚再逐步完善告警策略。毕竟一个能在凌晨三点准确把你叫醒的系统才是值得依赖的7×24小时预警系统。 爬虫7×24小时实时告警系统：企业微信与钉钉机器人接入实战
做爬虫的朋友大概率都有过这种经历凌晨两三点某个关键采集任务挂了一整夜第二天早上打开后台——企业微信和钉钉的告警群安安静静而数据库里该入库的数据一条没进来。最难受的还不是修代码而是你不知道它在几点挂掉的、中间发生了什么。给爬虫配上监控告警让异常在发生的第一时间变成手机上的实时预警这件事我是被几次深夜事故逼着认真做起来的。这篇文章就围绕怎么结合企业微信或钉钉打造一套爬虫专用的7×24小时实时预警系统来讲。我会把告警渠道选型、机器人接入细节、监控指标怎么定、告警触发策略怎么设计以及一套可以直接改改就用的Python告警模块全部梳理清楚。不管你在跑的是requests脚本、Scrapy项目还是分布式采集任务把这里的思路复制到自己的环境里基本都能用得上。1. 深夜静默失败我决定给爬虫装告警的那个夜晚1.1 爬虫悄无声息挂掉的几种典型死法爬虫不会像Web服务那样挂了就返回500。它最大的特点是程序还活着但采集质量已经严重下降了。我总结了几种最常见的静默死亡场景你可以对照一下自己有没有遇到过。第一种是目标站点页面改版。这是最普遍的死法。选择器失效、字段位置变动、翻页参数变化程序只负责发请求和解析页面结构一变解析结果直接变0。重点是这个过程中请求成功率依然很高日志也不会报错看起来一切正常实际上已经白跑了一整夜。第二种是代理IP池失效。做采集的基本都离不开代理尤其是对访问频率敏感的目标源。代理池里的IP一个个失效程序不断重试最终大量请求积压超时。有些任务配置了重试机制碰到失败就一直重试结果一个错误请求被放大成几十次日志刷屏也不报警。第三种是触发了反爬机制但程序没识别出来。服务器返回200Response内容却是验证码、跳转页或风控提示。如果解析逻辑里没有对这类内容做判断程序就会把垃圾页面当成正常数据来源解析字段全部为空。这种故障隐蔽性极高也是最难排查的。第四种是基础设施问题。磁盘写满、内存不足、服务被系统OOM杀掉或者分布式爬虫的某个Worker节点悄悄断连。任务队列里的消息没有消费者整个流程堵在那里但主进程还在运行看起来一切正常。第五种是外部依赖异常。比如目标数据源需要登录态而登录态的Cookie过期了或者爬虫依赖的某个接口调整了鉴权方式导致调用直接403。这类问题往往发生在凌晨因为很多站点会在低峰期更新服务器配置。1.2 没有告警时损失是被一步步放大的没有监控告警的时候一次静默故障的实际损失远不止几小时的数据缺失。第一步是数据断档。很多采集任务对时效性要求很高比如商品价格、股票行情、舆情信息错过的时间窗口补不回来。就算允许补采几小时的数据积压也够你追很久。我曾经有一个任务挂了6个小时恢复之后跑了整整一天才把积压数据补完期间新数据又要继续采两头挤压。第二步是信任损耗。爬虫跑出来的数据往往是给业务方、给下游系统用的。数据延迟一次人家能理解延迟多了整个采集系统的口碑就没了。每次都要你主动解释昨晚又挂了这种体验真的很消耗个人信誉。第三步是封禁风险被放大。很多情况下故障往往伴随着对目标站点的异常请求模式。比如代理失效后重试风暴或者登录态失效后反复请求异常接口这些行为都会加重风控识别。本来及时发现、及时停手就能避免的封禁因为没有告警反而让反爬系统看到了一个死而不僵的客户端处理起来更麻烦。所以告警系统不是为了在出事的时候通知一声那么表面它的本质是缩短故障发现时间。能把几个小时的故障发现时间压缩到几分钟损失的放大效应就被控制住了。这也是为什么我一直坚持爬虫的监控告警不是可选项它是整套采集系统里和爬虫代码同等重要的一部分。2. 告警渠道选型为什么最后留下了企业微信和钉钉2.1 邮件、短信和IM在告警场景下的实际体验差异告警消息要送到人手上渠道很关键。我最初用的是邮件后来又试过短信最后才把重心放到企业微信和钉钉这类IM机器人上。这里面的取舍很有意思。渠道到达时效格式能力成本被打断感适合场景邮件分钟级且容易进垃圾箱富文本强但手机上查看并不及时低很弱日报汇总、非紧急通知短信秒级但常有延迟和丢失纯文本无法排版按条收费很强电话无人接听时的兜底IM机器人秒级支持Markdown、引用、人免费可控实时告警、协作处理邮件的核心问题不是送不到而是送达不等于看见。人不会一直盯着邮箱尤其是凌晨。短信虽然能推但纯文本的格式太受限一条长告警发过来根本看不完。而企业微信和钉钉机器人几乎是专为这类场景设计的免费、秒达、能发结构化消息还能在群里责任人。另外说一句现在很多监控系统比如夜莺这类开源工具的告警通知也默认支持企业微信和钉钉机器人这说明IM渠道已经是社区验证过的主流方案了。你要做的不是从零发明而是把这类通知能力接入到自己的爬虫监控体系里。2.2 企业微信机器人适合已经用微信协作的团队企业微信机器人最大的优势是离人近。国内很多团队的工作沟通就在企业微信上告警消息推到群里大家顺手就看到了不需要额外打开App。从技术角度看企业微信群机器人的接入也足够省事在一个群里添加自定义机器人拿到Webhook地址然后往这个地址POST一个JSON消息就行。消息格式支持文本、Markdown、图片、图文链接等。Markdown对告警来说完全够用标题、加粗、链接、引用都能渲染。它的另一个特点是支持在消息里人。你可以通过mentioned_list传用户的userid也可以通过mentioned_mobile_list传手机号来指定的人。这个能力在告警分级里非常实用P1故障直接值班工程师而不是所有人手机都在响。限制方面企业微信群机器人官方对单个机器人有频率限制一般认为是每分钟20条左右。单看这个数字感觉不小但如果告警策略没做好凌晨一小时触发几百条直接触发限流事小把告警群变成垃圾群导致大家屏蔽通知才是大问题。这个我在后面的告警策略部分会重点讲。2.3 钉钉机器人安全设置更硬核钉钉机器人放在同一维度比较接入流程和企业微信很像建群、加自定义机器人、拿Webhook、POST消息。但它有几个差异点值得注意。首先是安全设置更丰富。钉钉自定义机器人必须选择一种安全设置自定义关键词、加签、IP白名单三种可以多选。自定义关键词要求推送的消息里必须包含指定关键词否则拒绝发送加签要求对请求做HMAC-SHA256签名IP白名单则只允许指定来源IP调用。相比企业微信钉钉的这套机制对防止Webhook地址泄露后被人乱刷的保护更完善。其次是消息类型。钉钉机器人支持text、markdown、link、actionCard、feedCard等类型。其中markdown类型同样支持标题、加粗、引用、列表告警场景完全够用。而actionCard类型可以推送一张带有按钮的卡片点击按钮跳转到指定链接这个功能用来做点击处理告警很顺手。选企业微信还是钉钉其实不用太纠结。看你们团队日常用哪个办公IM就选哪个如果两边都在用也可以同时接入做成双通道冗余。比如企业微信为主钉钉为备份通道这样万一一个IM服务出问题告警还能从另一个渠道送达。3. 企业微信机器人接入建群、Webhook、加签校验一条龙3.1 建群添加机器人的关键步骤与权限坑实际接入企业微信机器人差不多五分钟就能搞定。操作路径是在企业微信里创建一个群聊哪怕只有你自己一个人也行然后进入群设置找到群机器人选择添加机器人再选新创建一个机器人。给它起个名字比如爬虫告警创建成功后就能看到Webhook地址。这里面有几个坑我一个个说。第一只有群主和群管理员才有权限添加机器人。如果你在一个别人的群里没有管理权限是找不到添加机器人入口的。建议每个项目单独建一个告警群并由实际负责维护的人做群主。第二机器人创建后Webhook地址里包含一个key参数这个key就相当于机器人的密钥。不要把它写进会被提交到Git仓库的配置里更不要截图发到公开渠道。我的习惯是放到环境变量或独立的配置文件中并且配置文件的读取权限设成只有启动用户可读。第三如果机器人被移出群聊Webhook地址会失效所有调用都会报错。碰到这种情况重新创建一个机器人即可代码里更新一下地址。创建机器人的时候还可以选择加签安全设置。企业微信新版机器人支持配置一个签名密钥发送请求时需要在URL里带上计算的签名。这块我下一步详细说。3.2 Webhook消息格式与加签校验代码企业微信机器人发送消息时HTTP请求的Body是一个JSON。以Markdown消息为例最基本的格式是这样{ msgtype: markdown, markdown: { content: ## 爬虫预警\n 任务price_spider\n错误信息连续3个周期解析结果为空 } }如果配置了加签需要在Webhook地址后面拼接timestamp和sign两个参数。加签的算法逻辑如下取当前时间的Unix秒级时间戳拼字符串timestamp \n secret以secret为密钥对上述字符串做HMAC-SHA256摘要将摘要结果做Base64编码得到sign。下面是一段完整的Python实现包含加签URL生成和消息发送import base64 import hashlib import hmac import time import urllib.parse import requests class WecomNotifier: def __init__(self, webhook_url: str, secret: str ): self.webhook_url webhook_url self.secret secret def _signed_url(self) - str: if not self.secret: return self.webhook_url timestamp str(int(time.time())) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign base64.b64encode(hmac_code).decode(utf-8) # 实测发现 base64 结果里的 / 直接拼进 URL 可能被截断建议做一次编码 sign urllib.parse.quote_plus(sign) connector if ? in self.webhook_url else ? return f{self.webhook_url}{connector}timestamp{timestamp}sign{sign} def send_markdown(self, content: str) - bool: payload { msgtype: markdown, markdown: {content: content}, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: print(fsend failed: {result}) return False return True调用方式很简单wecom WecomNotifier( webhook_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyYOUR_KEY, secretYOUR_SECRET, ) wecom.send_markdown(## 爬虫预警\n 采集任务异常请检查。)实测经验签名校验失败时企业微信会返回类似invalid sign的提示。最容易踩的坑是服务器时间不同步导致本地生成的timestamp和服务端时间差太多签名直接不通过。上线前先确认运行告警模块的服务器时间是用NTP同步过的。3.3 企业微信推送的限流和格式限制企业微信机器人的限制主要是频率和Markdown语法范围。频率上单个机器人每分钟大概能推送20条左右超过后服务端会返回限流错误。这个限制对正常告警来说够用但前提是不能在告警策略里做每5秒推一条这种设计。Markdown格式方面企业微信支持的是其特定子集#标题、**加粗、引用、[文本](链接)、无序列表、有序列表、font color颜色设置等。不支持的语法会自动降级显示。实测中|表格语法在企业微信里是不会渲染成表格的这点和钉钉有差异。所以我在写告警模板时习惯用标题加引用的方式排版而不是用表格。推送内容上我建议一条消息不要超过2000字。手机端消息太长会被折叠反而抓不住重点。告警内容要摘要化什么任务、什么级别、什么现象、谁负责、处理入口这几个要素就够了细节留给日志系统去查。4. 钉钉机器人接入加签、关键词、IP白名单的取舍4.1 三种安全设置怎么选钉钉自定义机器人的安全设置可以单选或多选。三种方式的适用场景差别很大我列个表对比一下安全方式原理优点缺点推荐场景自定义关键词消息文本必须包含指定关键词配置简单关键词被猜中后可被伪造需要消息文本包含关键词内网告警追求低成本接入加签请求URL带HMAC签名安全性最高无法简单伪造需要在代码里实现签名逻辑生产环境默认推荐IP白名单只允许指定来源IP调用Webhook直接屏蔽外部IP如果服务器IP变化或走代理需要同步维护固定服务器部署我的建议很明确只要不是临时测试一律用加签。加签相当于给Webhook地址加了一把只有你的代码能打开的锁即使地址泄露别人也推不了消息。IP白名单可以当作辅助手段叠加使用但如果你的爬虫部署在多个动态IP节点上维护成本会比较高。4.2 钉钉Markdown消息格式与发送代码钉钉的Markdown消息格式和企业微信有些区别。钉钉机器人对Markdown的支持标题用####这种四级标题效果比较好看引用、**加粗、-列表都是支持的但不支持表格链接语法支持。下面是一段完整的钉钉加签发送实现import base64 import hashlib import hmac import time import urllib.parse import requests class DingTalkNotifier: def __init__(self, access_token: str, secret: str ): self.access_token access_token self.secret secret self.base_url https://oapi.dingtalk.com/robot/send def _signed_url(self) - str: timestamp str(round(time.time() * 1000)) # 注意钉钉用毫秒时间戳 url f{self.base_url}?access_token{self.access_token} if self.secret: string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) url ftimestamp{timestamp}sign{sign} return url def send_markdown(self, title: str, text: str, at_mobiles: list None) - bool: payload { msgtype: markdown, markdown: { title: title, text: text, }, } if at_mobiles: payload[at] { atMobiles: at_mobiles, isAtAll: False, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: print(fsend failed: {result}) return False return True用法ding DingTalkNotifier( access_tokenYOUR_ACCESS_TOKEN, secretSECYOUR_SECRET, ) ding.send_markdown( title爬虫预警, text#### 爬虫预警\n\n 任务price_spider\n\n**状态**连续3个周期解析结果为空\n\n[查看详情](https://your-monitor.example.com), at_mobiles[13800138000], )这里特别提醒一下钉钉加签的时间戳是毫秒级不是秒级。我之前用秒级时间戳签名验证就没通过排查了好久才发现是单位搞错了。另外加签后生成的sign必须用quote_plus编码因为Base64编码结果里可能包含、/、这些URL特殊字符不编码会被服务端解析错误。4.3 钉钉限流、错误码和重试策略钉钉自定义机器人的限流策略也比较严格。单个机器人每分钟最多推送大约20条如果调用过频繁服务端会返回限流错误。另外钉钉还会校验消息内容中是否包含匹配的安全设置比如你选择了关键词方式但消息里没有关键词会返回类似keywords not in content的错误错误码一般是310000一类的提示。处理这类失败时我建议在发送函数里做重试但要加退避。愚蠢的重试方式是失败后立刻再发一次这在凌晨告警场景里很可能把服务端限流打到更严重。我用的策略是第一次失败后等待2秒重试再失败就等5秒最多重试3次仍失败则放弃本次推送并把告警写入本地文件。宁可少推送一次也不要因为重试风暴把群里的消息渠道彻底搞挂。另外钉钉机器人发送失败并不都是网络问题也可能是消息格式不合法。这类问题重试多少遍都没用需要检查返回的errmsg提前在本地拦截。比如Markdown格式里的语法错误、文本长度超过限制这些在发送前就可以规避。5. 监控指标设计先定义值得半夜叫醒你的异常5.1 从日志中能提炼出的核心指标告警系统不能什么都报否则跟不报没区别。要设计好告警首先是定义监控指标。我把爬虫运维里最常用的核心指标整理成了下面这张表指标含义异常场景示例请求成功率成功请求数 / 总请求数代理失效、目标接口临时故障有效解析率成功解析出目标字段的页面占比页面改版、数据结构变更提取数据量单位时间窗口内入库的记录数数据积压、采集逻辑阻塞平均响应时间 / P95请求耗时的集中趋势和长尾情况带宽问题、目标源限速代理池可用率可用代理数 / 总代理数代理供应商故障、账号欠费队列积压量待采集任务队列里的消息数量Worker节点断连、消费能力下降进程存活状态爬虫主进程是否持续运行内存溢出、崩溃退出磁盘与内存使用率主机基础资源水位日志吞盘、数据未及时归档这些指标听起来多实际上都很好采集。最简单的做法是让爬虫把每次请求的关键信息写进结构化日志JSON格式的一行然后由监控程序定时读取日志窗口做聚合。稍微复杂一点的可以把指标上报到Prometheus再用Grafana展示。但告警模块本身不一定要依赖整套监控系统——对大多数脚本型爬虫来说一个能跑定时任务的Python进程就够了。5.2 单指标容易误判联动才是关键设计监控指标最容易犯的错误是只看单一指标。我来举个例子请求成功率下降了但有效解析率和数据量都正常。这大概率只是个别接口波动不影响大局不需要大呼小叫。反过来请求成功率很高但有效解析率是0这才是真正要命的故障——说明页面结构变了或反爬页面混进了正常请求里。所以我做告警判断时基本都会采用组合条件。比如连续3个统计周期每个周期5分钟内有效解析率低于50%并且提取数据量低于正常基线的30%才触发解析异常告警。任务队列积压量超过阈值同时活跃Worker数低于预期才触发分布式节点故障告警。请求成功率低于90%并且平均响应时间超过基线的两倍才触发网络链路异常告警。组合条件的逻辑很简单单个指标的波动可能有很多原因但多个指标同时反常往往意味着真实故障。这样做还有一个额外的好处是可以大幅减少误报——误报多了人就会疲于应对最后连真告警也没人看。5.3 阈值不要拍脑袋先跑一周基线阈值怎么定是最考验经验的地方。我的建议很简单不要在系统刚上线的时候拍脑袋定阈值先让它“只记录、不告警”地跑上一周把正常运行的指标分布收集起来再回头定阈值。比如页面解析成功率正常情况可能是99.5%左右偶尔跌到90%但很快恢复。如果你直接把阈值定在99%那这一周的指标会让你发现原来某些低峰时段本来就会有一些波动。合理的阈值应该设在“正常波动范围之外但又不至于等故障扩大到不可挽回才触发”的位置。具体方法上我个人喜欢用百分位数。比如提取数据量取过去一周每个5分钟窗口的P5和P95那么低于P5-20%或高于P9520%都可以视为异常候选。这种方法比单纯设一个绝对数字更贴近实际。当然每个业务不一样绝对阈值也有用比如磁盘使用率超过85%这类资源指标就和业务量无关直接定绝对值就行。6. 告警触发策略静默期、恢复通知、分级升级的设计6.1 阈值告警和趋势告警要怎么配合告警触发策略如果是超过一个固定阈值就推消息实现简单但容易漏报慢变故障。比如代理池可用率从80%慢慢降到30%过程中可能好几个小时都没有一次触发阈值等你发现时代理池已经崩了。所以在实践里我会把阈值告警和趋势告警结合起来用。趋势告警的方式不复杂每个统计周期都计算当前指标与历史基线的偏离程度一旦偏离持续扩大即使还没到绝对阈值也先发一条低级别告警提示。举个例子请求成功率当前98%还没到95%的硬阈值但和过去三天的同期水平相比下降了5个百分点并且连续两个周期都在下滑这时就应该推一条P2告警让人提前介入。这样做的好处是很多故障被消灭在刚冒头的阶段。代价是趋势告警需要有一个稳定的历史基线数据所以基线收集这一步不能跳过。6.2 静默期和去重防止凌晨的告警风暴告警系统的最大敌人是告警风暴。凌晨2点一个故障触发如果策略写得粗可能每隔两分钟就推一条相同告警。结果就是手机一晚上响个不停第二天群里全是重复消息所有人对告警产生免疫真正的严重故障反而被淹没。解决思路就是静默期和去重。静默期指的是同一类告警在指定时间内只允许推送一次比如600秒内同一个task_id的解析异常不再重复推送去重则是把多条同类异常聚合成一条消息附带最近10分钟内累计失败N次这样的汇总信息。我实现的告警管理器里会维护一个字典记录每个告警key的最后推送时间只有超过静默期才允许再次推送。具体的实现代码我会在下一章给出。这里先强调一个原则告警的目的是让人知道出了事和事有多严重不是让系统实时播报故障的每一个细节。细节属于日志系统和监控面板。6.3 分级与升级规则让通知更有轻重缓急告警必须分级不然后果就是P1和P3用同一种方式轰炸所有人。我常年的分级维度如下P1严重采集任务已停止、关键数据源不可用、代理池整体失效、磁盘满导致无法写入。这一类告警应立即推送并值班人。P2主要指标明显偏离但业务未完全中断比如解析率下降、响应时间飙升、队列积压增长。这类告警推送即可不需要所有人但需要人工确认。P3次要单次超时、夜间低峰期的一些小波动、个别页面解析失败。这类场景不建议实时推送而是汇总成日报白天统一看。升级规则也很重要。我的策略是P3持续超过1小时自动升级为P2P2持续超过30分钟仍未恢复自动升级为P1并追加人。升级的本质是给故障一个被处理的机会人不回不看告警就会自己往更高层级走直到有人接手。7. 实战代码一个同时兼容企业微信和钉钉的告警模块7.1 通用告警消息结构定义既然企业微信和钉钉都要接入那就先把告警消息抽象成一个数据结构收发双方都依赖这个结构。这样可以保证切换渠道时告警内容的组装逻辑不用改动。我用dataclass定义一个AlarmMsg它包含标题、内容、级别、任务ID、对象和时间戳等字段from dataclasses import dataclass from typing import List, Optional dataclass class AlarmMsg: title: str content: str level: str P2 # P1 / P2 / P3 task_id: str at_mobiles: Optional[List[str]] None # 钉钉 手机号 at_userids: Optional[List[str]] None # 企业微信 userid timestamp: str 然后在组装消息时统一由这个对象出发企业微信和钉钉的Notifier各自负责把它渲染成对应平台的Markdown格式。这样做的好处是以后就算要加一个飞书机器人也只需要再写一个Notifier类业务代码几乎不用动。7.2 企业微信Notifier实现结合前面讲到的加签逻辑我把企业微信的Notifier封装成一个类import base64 import hashlib import hmac import time import urllib.parse import requests class WecomNotifier: def __init__(self, webhook_url: str, secret: str ): self.webhook_url webhook_url self.secret secret def _signed_url(self) - str: if not self.secret: return self.webhook_url timestamp str(int(time.time())) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) connector if ? in self.webhook_url else ? return f{self.webhook_url}{connector}timestamp{timestamp}sign{sign} def send(self, msg: AlarmMsg) - bool: text f## {msg.title}\n text f 级别**{msg.level}**\n if msg.task_id: text f 任务{msg.task_id}\n text f时间{msg.timestamp}\n\n text msg.content payload { msgtype: markdown, markdown: {content: text}, } if msg.at_userids: payload[mentioned_list] msg.at_userids resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() return result.get(errcode) 0这里有个细节企业微信的mentioned_list需要传成员UserID如果你的账号体系里拿不到UserID可以用mentioned_mobile_list传手机号两者效果类似。7.3 钉钉Notifier实现钉钉的Notifier实现逻辑很接近区别在于时间戳用毫秒、签名要拼在access_token后面class DingTalkNotifier: def __init__(self, access_token: str, secret: str ): self.access_token access_token self.secret secret self.base_url https://oapi.dingtalk.com/robot/send def _signed_url(self) - str: timestamp str(round(time.time() * 1000)) url f{self.base_url}?access_token{self.access_token} if self.secret: string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) url ftimestamp{timestamp}sign{sign} return url def send(self, msg: AlarmMsg) - bool: text f#### {msg.title}\n\n text f 级别**{msg.level}**\n if msg.task_id: text f 任务{msg.task_id}\n text f时间{msg.timestamp}\n\n text msg.content payload { msgtype: markdown, markdown: { title: msg.title, text: text, }, } if msg.at_mobiles: payload[at] { atMobiles: msg.at_mobiles, isAtAll: False, } resp requests.post(self._signed_url(), jsonpayload, timeout10) result resp.json() return result.get(errcode) 0钉钉的消息text里不支持企业微信里那种##一级标题的渲染效果我实测####展示出来更干净。这个差异属于平台特性写模板时要分开处理。7.4 集成到爬虫主循环监控-判定-发送的完整链路有了Notifier还差一个控制静默期的管理器。我用一个简单的类来实现去重逻辑import time class AlarmManager: def __init__(self, notifier, cooldown_seconds: int 600): self.notifier notifier self.cooldown_seconds cooldown_seconds self._last_sent {} def should_alarm(self, key: str) - bool: now time.time() last self._last_sent.get(key, 0) if now - last self.cooldown_seconds: return False self._last_sent[key] now return True def send(self, msg: AlarmMsg, key: str ) - bool: key key or msg.title if not self.should_alarm(key): return False return self.notifier.send(msg)然后把它接到爬虫主循环里。下面是一个简化但完整的示例演示了抓取-计算指标-判断异常-告警的全链路def crawl_once(): # 这里替换成你的真实采集逻辑 stats { request_success_rate: 0.98, parse_success_rate: 0.30, item_count: 120, task_id: price_spider, } return stats def run_monitor(): notifier WecomNotifier(WECOM_WEBHOOK_URL, WECOM_SECRET) alarm AlarmManager(notifier, cooldown_seconds600) while True: try: stats crawl_once() # 组合条件判定解析率低 数据量少 - P1 if stats[parse_success_rate] 0.5 and stats[item_count] 300: alarm.send( AlarmMsg( title采集任务疑似解析失效, content- 有效解析率{:.2%}\n- 提取数据量{}\n- 建议检查目标站点页面结构。.format( stats[parse_success_rate], stats[item_count] ), levelP1, task_idstats[task_id], timestamptime.strftime(%Y-%m-%d %H:%M:%S), ), keyfparse-fail:{stats[task_id]}, ) except Exception as exc: alarm.send( AlarmMsg( title采集任务异常退出, contentf异常信息{exc}, levelP1, task_idprice_spider, timestamptime.strftime(%Y-%m-%d %H:%M:%S), ), keyspider-exception, ) time.sleep(30) if __name__ __main__: run_monitor()这个环节里最核心的部分就是AlarmManager的静默期判断。没有它一个故障在600秒内反复触发20次你的手机就会被消息淹没。有了它同类告警每10分钟最多推一条既不会漏也不会吵。8. 部署与运维让监控系统自己也要24小时在线8.1 用systemd守护爬虫和监控进程监控模块本身也是一个进程它也有挂掉的风险。所以我会用systemd来守护爬虫任务和告警监控进程而不是简单地在终端里nohup一下。下面是我常用的一个systemd unit配置示例[Unit] DescriptionSpider Alarm Monitor Afternetwork-online.target [Service] Userspider WorkingDirectory/data/spider ExecStart/usr/bin/python3 /data/spider/alarm_monitor.py Restartalways RestartSec10 EnvironmentFile/data/spider/.env [Install] WantedBymulti-user.target配置好以后依次执行systemctl daemon-reload、systemctl enable spider-alarm、systemctl start spider-alarm就能开机自启。Restartalways是关键进程异常退出后10秒内会自动拉起。环境变量通过EnvironmentFile引入避免把密钥直接写在代码里。8.2 给监控系统加心跳长时间没消息就是最大的告警监控系统最尴尬的一个坑是监控进程挂了之后大家都不知道。因为没人收到告警这件事本身不会产生告警。我解决这个问题的方法是给监控系统加心跳机制。最简单的方式是每天固定时间在告警群里推一条汇总消息内容包括今日任务总数、成功任务数、失败任务数、告警触发次数。这条消息本身就是一个健康信号——如果某天早上没收到日报说明监控进程可能挂了。更进一步的做法是在多个核心采集节点上部署互看的探针。比如A节点的监控模块定时向B节点的Webhook地址发一个存活探测请求B节点收到后记录如果连续几次没收到A的心跳B就替A发出一条告警。这套互看机制不复杂但对监控系统自身的可用性能带来很大的提升。8.3 几个隐蔽但致命的运维坑最后分享几个我实际运维中踩过或帮别人排查过的坑都很隐蔽但遇到了都挺折腾。第一个是服务器时间不同步。前面已经提过一次这里再强调无论企业微信还是钉钉加签都依赖请求时间与服务器时间接近。如果服务器时间偏差超过一两分钟签名校验就会失败。我一般会在服务器上配置NTP自动同步并在告警模块启动时先校验一次本机时间。第二个是Webhook地址泄露。有一次我发现告警群里莫名其妙出现一些不是我写的测试消息排查后发现是Webhook地址被同事贴到公开文档里了。处理办法有两种在企业微信端删掉机器人重新创建一个在钉钉端开启IP白名单把无关来源直接挡在门外。平时养成习惯Webhook地址和secret只放进部署服务器的环境变量里。第三个是告警模块与爬虫部署在同一台机器磁盘满时一起挂。爬虫最占磁盘的往往是日志和抓取缓存一旦把盘写满爬虫和监控进程都会因为写日志失败而异常。我会把监控模块的日志单独写到另一个分区同时给日志目录配置按大小轮转避免单个日志文件无限增长。第四个是手机端的通知权限。企业微信和钉钉的群消息默认是免打扰的如果告警群没有开启关注的消息提醒消息倒是送到了但手机状态栏不弹提醒等于白搭。所以建完告警群第一件事就是把群设置里的消息通知改成关注的人或指定消息提醒并把机器人的消息类型设为关注项然后拿测试消息实际验证一遍手机能不能响。这些坑单看都不大但叠加在一起就足以让一套架构上看起来很完美的告警系统在关键时刻变成了摆设。我的经验是告警系统上线之后不急着加更多指标先花一周时间把消息能否稳定到达手机这件事反复验证清楚再逐步完善告警策略。毕竟一个能在凌晨三点准确把你叫醒的系统才是值得依赖的7×24小时预警系统。